MCP 도구

MCP 도구 (Tools)

MCPAdapter는 MCP 서버와 LangChain 에이전트를 이어주는 다리 역할을 해요. 서버가 광고하는 도구를 발견해 표준 LangChain 도구로 변환하죠. list_tools()에서 받은 도구를 다른 LangChain 도구처럼 create_agent에 전달하면 됩니다. 이 페이지는 그 다리 특유의 부분 — MCP 도구 식별, 실행 제어, 출력 처리, 서버가 호출 중 입력을 요구할 때의 대응 — 을 다룹니다. 실행 가능한 발견·에이전트 예시는 MCP quickstart를 참고하세요.

출처: LangChain 공식 문서 — mcp/tools

⚠️ langchain.mcp 네임스페이스는 langchain[mcp]>=1.4.0이 필요하고 현재 베타 상태예요. API는 변경될 수 있어요.

에이전트에서 MCP 도구 사용하기 (Use MCP tools in an agent)

MCPAdapter.list_tools로 서버의 카탈로그를 발견하고, 반환된 도구를 create_agent에 전달해요. 에이전트 관점에서는 이들이 LangChain 도구처럼 동작합니다 — 모델이 도구를 선택하고, LangChain이 호출하며, 결과 ToolMessage가 모델로 돌아가요.

from langchain.agents import create_agent
from langchain.mcp import MCPAdapter

async def run_agent(server) -> dict:
    # Discover the server's tools, then hand them to the agent like any other
    # LangChain tools. The tools hold the client, so the agent stays usable
    # for the life of the adapter context.
    async with MCPAdapter(server) as adapter:
        tools = await adapter.list_tools()
        agent = create_agent("claude-sonnet-5", tools)
        return await agent.ainvoke(
            {
                "messages": [
                    {"role": "user", "content": "What is the forecast for Oslo?"}
                ]
            }
        )

LangChain 도구의 정의·바인딩·사용에 대한 일반적인 지침은 Tools를 참고하세요. 여러 MCP 서버와 네임스페이스된 도구 카탈로그는 Connections을 보세요.

도구 메타데이터 (Tool metadata)

변환된 각 도구는 LangChain 도구의 메타데이터에 mcp 네임스페이스로 MCP 출처 정보를 담을 수 있어요.

tool.metadata
# {
#     "mcp": {
#         "tool": {
#             "annotations": {
#                 "destructive_hint": True,
#                 "read_only_hint": False,
#             },
#             "_meta": {"origin": "crm"},
#         },
#         "server": {
#             "name": "crm",
#             "version": "2.1.0",
#         },
#     },
# }

중첩된 각 필드는 선택적이에요. 서버는 도구 주석, _meta, 서버 신원, 이들의 조합, 또는 아무것도 제공하지 않을 수 있어요. annotations에는 read_only_hint, destructive_hint 같은 MCP 힌트가 들어 있고, _meta는 서버가 제공하는 불투명한 메타데이터이며, server는 그 도구를 광고한 MCP 구현체를 식별합니다.

선택적 메타데이터는 방어적으로 읽어야 해요. 필드가 없으면 예외를 일으키는 대신 기본값을 반환하게요.

from langchain.tools import BaseTool

def is_destructive(tool: BaseTool) -> bool:
    """Read the MCP destructive hint off the adapter's tool metadata."""
    # Chain `.get` with defaults so a tool missing any nested field returns
    # False rather than raising.
    annotations = (
        (tool.metadata or {}).get("mcp", {}).get("tool", {}).get("annotations", {})
    )
    return annotations.get("destructive_hint", False)

도구 출력 처리하기 (Handle tool outputs)

MCP 도구 결과는 LangChain 네이티브 값이 돼요 — 모델이 읽을 수 있는 콘텐츠, 구조화된 출력을 위한 artifact, 그리고 서버가 보고한 오류와 트랜스포트 실패를 구분하는 ToolMessage 상태까지요.

멀티모달 콘텐츠 (Multimodal content)

MCP 도구 결과는 LangChain 콘텐츠 블록으로 도착해요. 이미지·파일 콘텐츠는 text와 나란히 표준화된 image·file 블록으로 변환됩니다. 스크린샷을 반환하는 도구가 이미지 블록으로 모델에 도달하는 식이죠.

from langchain.mcp import MCPAdapter

async def access_multimodal_tool_content(server) -> None:
    async with MCPAdapter(server) as adapter:
        [screenshot] = await adapter.list_tools()

    # An MCP result arrives as LangChain content blocks. Image and file content
    # convert into standardized `image`/`file` blocks alongside `text`.
    message = await screenshot.ainvoke(
        {"name": "take_screenshot", "args": {}, "id": "1", "type": "tool_call"}
    )
    for block in message.content_blocks:
        if block["type"] == "text":
            print(f"Text: {block['text']}")
        elif block["type"] == "image":
            print(f"Image mime type: {block.get('mime_type')}")
            print(
                f"Image base64: {block.get('base64', '')[:20]}..."
            )

구조화된 콘텐츠 (Structured content)

도구가 구조화된 콘텐츠를 반환하면, 어댑터는 그것을 모델이 보는 텍스트에 섞지 않고 ToolMessage에 artifact로 붙여요. message.artifact에서 읽으면 됩니다.

message = await tool.ainvoke({"name": "...", "args": {...}, "id": "1", "type": "tool_call"})
if message.artifact is not None:
    structured = message.artifact["structured_content"]

artifact는 MCPToolArtifact이며, 그 structured_content 필드가 도구 결과의 structuredContent를 담아요.

오류 (Errors)

MCP 도구 결과는 isError 플래그를 담아요. 서버가 isError=True를 보고하면 어댑터는 그것을 서버의 메시지를 담은 status="error"ToolMessage로 변환하므로, 에이전트가 읽고 스스로 바로잡을 수 있어요.

from langchain.mcp import MCPAdapter

async def divide_by_zero(server):
    async with MCPAdapter(server) as adapter:
        [divide] = await adapter.list_tools()

    # A server error (isError=True) reaches the model as a failed ToolMessage,
    # so the agent can read the server's own message and retry. Transport
    # failures still raise, because a model cannot act on those.
    return await divide.ainvoke(
        {"name": "divide", "args": {"a": 10, "b": 0}, "id": "1", "type": "tool_call"}
    )

서버가 보고한 오류는 실패한 도구 메시지로 모델에 도달하지만, 트랜스포트·세션 실패는 예외를 일으켜요. 모델은 끊긴 연결에 대해 조치할 수 없으니까요.

Human-in-the-loop

주석(annotation)을 읽으면 도구 이름을 하드코딩하는 대신, 서버가 도구에 대해 선언한 내용을 기준으로 도구를 게이트할 수 있어요. MCP 서버는 destructiveHint 주석으로 도구가 파괴적임을 표시할 수 있고, MCPAdapter는 이를 metadata["mcp"]["tool"]["annotations"]["destructive_hint"] 아래에 노출합니다.

InterruptOnConfigwhen 프레디킷을 주세요 — 대기 중인 ToolCallRequest를 받아 그 호출이 승인을 필요로 하는지 반환하는 콜러블이에요. 파괴적 힌트를 로드 시점에 한 번 메타데이터에서 읽고, 콜러블이 호출마다 결정하게 하면, 서버가 노출하는 어떤 파괴적 도구든 도구 이름을 하드코딩하지 않고 하나의 설정으로 커버합니다.

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.agents.middleware.human_in_the_loop import InterruptOnConfig
from langchain.tools import BaseTool
from langchain.tools.tool_node import ToolCallRequest

def is_destructive(tool: BaseTool) -> bool:
    """Read the MCP destructive hint off the adapter's tool metadata."""
    annotations = (
        (tool.metadata or {}).get("mcp", {}).get("tool", {}).get("annotations", {})
    )
    return annotations.get("destructive_hint", False)

async def gate_destructive_tools(server):
    async with MCPAdapter(server) as adapter:
        tools = await adapter.list_tools()

        # Read the destructive hint from metadata once, then let a callable
        # decide per call. One config covers whatever destructive tools a
        # server exposes, without hardcoding tool names.
        destructive = {tool.name for tool in tools if is_destructive(tool)}

        def needs_approval(request: ToolCallRequest) -> bool:
            return request.tool_call["name"] in destructive

        gate = InterruptOnConfig(
            allowed_decisions=["approve", "reject"], when=needs_approval
        )
        interrupt_on: dict[str, bool | InterruptOnConfig] = {
            tool.name: gate for tool in tools
        }
        return create_agent(
            "claude-sonnet-5",
            tools,
            middleware=[HumanInTheLoopMiddleware(interrupt_on=interrupt_on)],
            checkpointer=InMemorySaver(),
        )

에이전트가 프레디킷이 게이트하는 도구를 호출하면 실행이 멈춰요. 승인하면 도구를 실행하고, 거부하면 도구를 건너뛰고 모델에 알립니다.

from langgraph.types import Command

# Approve the pending destructive call and resume.
resumed = await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config)

프레디킷은 request.tool_call["args"]를 통해 호출의 인자도 볼 수 있어요. 그래서 도구가 안전한 입력에서는 자유롭게 실행되고 위험한 입력(보호된 경로를 대상으로 하는 delete_file 호출 같은)에서만 멈추게 할 수 있습니다. 두 가지를 조합하면 도구를 그 유형과 인자가 정당화할 때에만 게이트할 수 있어요. 전체 승인 워크플로는 Human-in-the-loop을 참고하세요.

도구 실행 중 서버 요청 (Server requests during tool execution)

대부분의 도구는 호출 중간에 클라이언트에게 아무것도 요구하지 않고 끝나요. 서버가 입력을 필요로 할 때, MCPAdapter는 LangGraph interrupt를 통해 elicitation에 자동으로 응답합니다.

Elicitation

Elicitation은 도구 호출 중간에 서버가 입력을 요청하는 MCP 메커니즘이에요. 서버가 입력을 필요로 하면, 요청이 LangGraph 인터럽트로 표면화되어 이미 에이전트 작업을 검토하고 있는 사람이 답하고 실행이 재개됩니다.

from typing import Any

from langchain.agents import create_agent
from langchain.mcp import MCPAdapter
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

async def book_with_elicitation(server) -> dict:
    # Elicitation is handled automatically: when a server needs input mid-call,
    # the adapter surfaces the question as a LangGraph `interrupt()`, so the
    # person already reviewing the agent's work answers it and the run resumes.
    async with MCPAdapter(server) as adapter:
        tools = await adapter.list_tools()

        # Resuming a paused run needs persistence, so the interrupted run has
        # somewhere to wait.
        agent = create_agent("claude-sonnet-5", tools, checkpointer=InMemorySaver())
        config: Any = {"configurable": {"thread_id": "booking-1"}}

        paused = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "Book a table for 4."}]}, config
        )
        [interrupt] = paused["__interrupt__"]
        [question] = interrupt.value["requests"]

        # Answers are keyed by the server's own request key, so nothing has to
        # be tracked across the pause. `decline` or `cancel` would refuse.
        answer = {"action": "accept", "content": {"date": "2026-09-14"}}
        return await agent.ainvoke(
            Command(resume={"responses": {question["key"]: answer}}), config
        )

몇 가지 주의할 점이 있어요.

  • Elicitation은 기본적으로 켜져 있어요. 어댑터는 자기가 만드는 모든 클라이언트에 그 능력을 광고하고 인터럽트 루프를 구동해요. 자체 elicitation 핸들러를 이미 들고 있는 사전 빌드 클라이언트는 덮어쓰지 않고 존중됩니다.
  • 재개에는 영속성이 필요해요. 체크포인터를 붙여서 중단된 실행이 기다릴 곳을 제공하세요.
  • 답변은 서버의 요청 키로 키잉됩니다. Command(resume={"responses": {key: answer}})로 재개하세요. 각 답변의 actionaccept(요청 스키마와 일치하는 content와 함께), decline(답변 거부, 호출은 계속), cancel(호출 전체를 포기) 중 하나예요.

인터럽트 페이로드와 답변 유형은 langchain.mcp.elicitation에 있습니다.

elicitation만 이런 방식으로 응답돼요. 대신 sampling(LLM 완성 실행)이나 roots(도달 가능한 로컬 경로)을 요청하는 서버는 NotImplementedError를 일으켜요 — 현대의 세션 없는 프로토콜에는 그런 요청을 위한 실시간 백채널이 없기 때문이에요. Sampling and roots를 참고하세요.

인터럽트 기반 elicitation은 요청을 InputRequiredResult(현대 프로토콜의 input-required 라운드)로 반환하는 서버에 응답합니다. 레거시 핸드셰이크 세션 위에서 elicitation을 밀어내기만 하는 서버는 이런 방식으로 응답할 수 없어요.

함께 보기 (See also)

더 알아보기 (Learn more)