MCP 도구

MCP 도구 (MCP Tools · LangChain Python)

MCP 도구를 LangChain 에이전트에 로드하고, 실행을 제어하고, 서버 결과와 요청을 처리해요.

MCPAdapter는 MCP 서버와 LangChain 에이전트를 이어 주는 브리지예요. 서버가 광고하는 도구를 발견해서 표준 LangChain 도구로 적응시켜 주죠. list_tools()에서 나온 도구를 다른 LangChain 도구처럼 create_agent에 넘겨 주면 돼요.

이 페이지에서는 그 브리지에 특화된 내용을 다뤄요: MCP 도구 식별, 실행 제어, 출력 처리, 호출 중 서버가 입력을 요청할 때 응답하기. 실행 가능한 발견·에이전트 예제는 MCP 퀵스타트를 보세요.

에이전트에서 MCP 도구 쓰기

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 도구를 정의·바인딩·사용하는 일반적인 지침은 도구를 보세요. 여러 MCP 서버와 그 네임스페이스 도구 카탈로그는 연결을 보세요.

도구 메타데이터

적응된 각 도구는 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)

도구 출력 처리

MCP 도구 결과는 LangChain 네이티브 값이 돼요: 모델이 읽을 수 있는 콘텐츠, 구조화된 출력용 아티팩트, 그리고 서버가 보고한 오류와 전송 실패를 구분하는 ToolMessage 상태.

멀티모달 콘텐츠

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:  # [!code highlight]
        if block["type"] == "text":  # [!code highlight]
            print(f"Text: {block['text']}")  # [!code highlight]
        elif block["type"] == "image":  # [!code highlight]
            print(f"Image mime type: {block.get('mime_type')}")  # [!code highlight]
            print(  # [!code highlight]
                f"Image base64: {block.get('base64', '')[:20]}..."  # [!code highlight]
            )  # [!code highlight]

구조화된 콘텐츠

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

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

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

오류

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

서버가 보고한 오류는 실패한 도구 메시지로 모델에 도달하지만, 전송·세션 실패는 예외를 던져요. 모델은 끊어진 연결에 대해 행동할 수 없기 때문이에요.

사람-인-더-루프

주석을 읽으면 도구 이름을 하드코딩하는 대신 서버가 도구에 대해 선언한 내용에 기반해 도구를 게이트할 수 있어요. MCP 서버는 destructiveHint 주석으로 도구를 파괴적이라고 표시할 수 있는데, MCPAdapter가 metadata["mcp"]["tool"]["annotations"]["destructive_hint"]로 표면화해요.

InterruptOnConfig에 when 술어를 주세요: 대기 중인 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 호출)에만 일시 중지할 수 있어요. 둘을 결합하면 도구의 유형과 인수가 모두 합당할 때만 도구를 게이트할 수 있어요.

전체 승인 워크플로는 사람-인-더-루프를 참고하세요.

도구 실행 중 서버 요청

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

Elicitation

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

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 핸들러를 가진 사전 빌드 클라이언트는 덮어쓰지 않고 존중해요.
• 재개에는 지속성이 필요해요. 인터럽트된 run이 대기할 곳이 있도록 체크포인터를 붙이세요.
• 답은 서버의 요청 키로 키잉돼요. Command(resume={"responses": {key: answer}})로 재개해요. 각 답의 action은 accept(요청 스키마와 일치하는 content 포함), decline(답 거절, 호출 계속), cancel(전체 호출 포기) 중 하나예요.

인터럽트 페이로드와 답 유형은 langchain.mcp.elicitation에 있어요.

이렇게 답하는 건 elicitation뿐이에요. 서버가 대신 sampling(LLM 완성 실행)이나 roots(도달 가능한 로컬 경로)를 요청하면 NotImplementedError를 던져요. 현대의 세션리스 프로토콜에는 그 요청들에 대한 라이브 백채널이 없기 때문이에요. Sampling과 roots를 보세요.

같이 보기

• 콘텐츠 블록
• 도구
• 사람-인-더-루프
• FastMCP 도구 호출
• FastMCP 클라이언트 elicitation
• FastMCP 서버 elicitation
• MCP elicitation 스펙
• MCP 도구 주석

출처: 공식문서 - MCP Tools