에이전트 아키텍처를 정리한 프로토콜

발행: 2시간 전 (2026년 6월 16일 AM 12:00 GMT+9)

11 분 소요

Source: Towards Data Science

몇 주 전 데이터 팀의 한 사람이 우리 복잡한 에이전트 시스템 중 하나의 도구에 의해 채워지고 있는 데이터베이스 스키마를 업데이트할 수 있는지 물어보았다. 업데이트는 간단하다: 테이블에 두 개의 새 컬럼이 추가되는 것이다.

도구 정의는 에이전트 오케스트레이터에 있었고, 유사한 두 번째 버전은 검증 에이전트에 존재했다. 약 three sprints 전 누군가에 의해 작성된 utility 모듈에 약간 다른 버전과 오래된 것이 있었다. 인간‑in‑the‑loop 승인 로직은 그래프 엣지에 직접 연결되어 있었으며, 도구당 커스텀 구현이 하나씩 존재했다. 스키마 변경을 하려면 네 개의 파일을 수정하고, 각 에이전트를 별도로 재테스트한 후 다운스트림에 아무 문제 없이 깨지지 않기를 바라는 것이었다.

우리는 이를 해결했지만 심각한 질문을 제기했다: 왜 우리는 이렇게 만들었을까?

정직한 답은 우리가 대체할 수 있는 옵션이 없었다는 것이다. LangGraph에서 도구 호출은 설계상 로컬 관심사다. 필요한 곳에 도구를 정의하고, 필요할 때 호출하며, 모든 파이프라인을 소유한다. 두 개의 에이전트만 있을 때는 관리하기 쉽지만, 일곱 개의 에이전트가 중복되는 도구를 공유하고 인간 게이트를 두는 상황에서는 문제가 발생한다.

조사를 진행한 후 우리는 로컬에 각 에이전트마다 도구를 정의하는 대신, 모든 에이전트가 사용할 수 있는 공유 자원을 사용하기로 결정했다.

What is MCP?

Model Context Protocol은 2024년 말 Anthropic에서 공개한 오픈 표준이다. 이 표준은 AI 에이전트가 도구를 검색하고 호출하는 방식을 표준화한다. 오케스트레이터 안에 도구를 정의하는 대신 별도 서버에서 실행한다. 에이전트는 런타임에 해당 서버에 연결하고, 사용 가능한 도구가 무엇인지 물으며, 목록을 받아온다.

Senior engineer가 이 기사를 읽고 즉시 할 질문은 “단순히 중앙 집중식 도구 레지스트리를 만들고 각 에이전트에 시작 시 주입하지 않을 수 있지?” 라는 것이다. 나는 스스로에게 이 질문을 던졌고, 다른 시스템에서는 MCP 대신 직접 만든 레지스트리를 사용했다.

네, 가능합니다. 이미 그러한 것이 작동 중이면 MCP가 급한 상황이 아닙니다. 맞춤형 레지스트리는 interoperability boundary(상호운용 경계)를 제공하지 않습니다. MCP는 라이브러리가 아니라 프로토콜입니다. MCP‑compatible 클라이언트가 서버에 연결할 수 있게 하며, LangGraph 오늘, 다음 해 다른 프레임워크도 가능하다고 할 수 있습니다. TypeScript 클라이언트가 Python 서버를 별도의 통합 작업 없이 호출할 수 있습니다. 도구 레지스트리는 이러한 기능을 제공하지 않습니다.

또한 팀 소유권 관점에서도 MCP는 장점이 있다. 우리 경우 ML 팀이 도구를, 애플리케이션 팀이 그래프를 담당했다. MCP는 공유 코드베이스 없이 깨끗한 계약을 제공했다.

Building the MCP Server

MCP 서버는 Tools(호출 가능한 작업), Resources(읽기 전용 데이터), Prompts(재사용 가능한 템플릿) 세 가지를 노출할 수 있다. 에이전트 시스템이 작업을 수행해야 하는 경우, 도구가 핵심 관심사다.

Python SDK에는 FastMCP가 포함되어 있으며, 타입 힌트에서 스키마를 생성하고 프로토콜 라이프사이클을 관리한다. 함수만 작성하고 @mcp.tool() 데코레이터로 장식하면 서버는 나머지를 처리한다.

stdio 전송으로 людей가 놀라는 한 가지 점은 stdout에 절대 쓰지 말 것이다. MCP 프로토콜은 stdout을 통신 채널로 사용한다. 임의의 print() 호출은 메시지 스트림을 매우 혼란스럽게 만들 수 있다.

import sys
import logging
from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO, stream=sys.stderr)
logger = logging.getLogger("analyst-tools")

mcp = FastMCP("analyst-tools")

@mcp.tool()
async def run_analysis(code: str, dataset: str) -> dict:
    """
    실행 가능한 파이썬 스니펫을 실제 데이터와 실행하고 결과를 반환합니다.
    사용자가 집계 계산, 레코드 필터링 또는 통찰 도출을 원할 때 사용합니다.
    최종 출력은 변수 이름 'output'에 할당되어야 합니다.

    Args:
        code: 실행할 파이썬 코드.
        dataset: 'sales', 'inventory', 'pipeline' 중 하나.
    """
    logger.info(f"run_analysis | dataset={dataset}")
    return await execute_in_sandbox(code, dataset)

@mcp.tool()
async def write_to_db(table: str, payload: dict) -> dict:
    """
    분석 결과 테이블에 레코드를 저장합니다.
    run_analysis가 검증된 출력을 반환한 후에만 호출하세요.

    Args:
        table: 대상 테이블 이름.
        payload: 새 레코드로 쓰기 위한 키-값 쌍.
    """
    logger.info(f"write_to_db | table={table}")
    return await persist_result(table, payload)

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

Docstring은 LLM이 에이전트가 어떤 도구를 호출할지 판단하도록 돕는다. 따라서 좋은 docstring을 쓰는 것이 매우 중요하다.

Stdio vs HTTP

이 결정을 production 배포에서 매번 마주하게 되며, 대부분의 기사는 이를 건너뛴다.

Stdio는 서버를 클라이언트의 서브프로세스로 실행한다. 통신은 표준 입력/출력(standard input and output)을 통해 이루어진다. 지연 시간은 단일 자리 밀리초이며 네트워크가 전혀 관여하지 않으며, 설정도 최소화된다. 로컬 개발, 단일 머신 배포, 혹은 서버와 클라이언트가 동일한 프로세스 트리 안에 있을 때 적합하다.

Streamable HTTP는 서버를 독립적인 서비스로 실행한다. 서버가 여러 클라이언트 또는 머신에 공유되어야 하거나 컨테이너로 배포하고 가로 확장(horizontal scaling)이 필요할 때 사용한다. Cloud Run과 같은 Serverless 배포도 잘 맞는다. Stdio는 장수 수명(parent process)이 필요한 모델이기 때문에 Serverless와 호환되지 않는다.

FastMCP에서 이 두 가지 전환은 한 줄만 변경하면 된다:

mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

데이터 주권 요구사항을 위해 온‑프레미스에 실행되는 MCP 서버와 외부 API를 전혀 호출하지 않는 도구를 사용하면 컴플라이언스 팀에게 깔끔한 이야기를 할 수 있다. 프로토콜은 서버가 어디에서 실행되는지 신경 쓰지 않는다.

Connecting it to LangGraph

langchain-mcp-adapters 라이브러리는 서브프로세스 생명주기 관리를 담당하고, 도구 검색 핸드쉐이크를 수행하며, MCP 도구 스키마를 LangChain‑호환 가능한 도체 객체로 변환한다.

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_google_vertexai import ChatVertexAI

llm = ChatVertexAI(
    model="gemini-2.5-flash",
    temperature=0,
    max_tokens=None
)

async def run(query: str):
    async with MultiServerMCPClient({
        "analyst-tools": {
            "command": "python",
            "args": ["./mcp_server.py"],
            "transport": "stdio",
        }
    }) as client:

        tools = await client.get_tools()
        llm_with_tools = llm.bind_tools(tools)

        def agent_node(state: MessagesState):
            return {"messages": [llm_with_tools.invoke(state["messages"])]}

        graph = StateGraph(MessagesState)
        graph.add_node("agent", agent_node)
        graph.add_node("tools", ToolNode(tools))
        graph.add_edge(START, "agent")
        graph.add_conditional_edges("agent", tools_condition)
        graph.add_edge("tools", "agent")

        app = graph.compile()
        result = await app.ainvoke({
            "messages": [{"role": "user", "content": query}]
        })

        print(result["messages"][-1].content)

tools_condition은 LangGraph에 내장된 모듈로, 마지막 메시지에 도구 호출이 포함돼 있는지 확인한다. 있다면 도구 실행기로 라우팅하고, 없으면 작업을 마무리한다. 자체 라우팅 함수를 작성하는 것보다 이 모듈을 사용하는 것이 더 안전하다 porque 엣지 케이스를 처리하고 구현 누락을 줄이기 때문이다.

One beha… (문장이 완성되지 않음)

에이전트 아키텍처를 정리한 프로토콜

In This Article

What is MCP?

Building the MCP Server

Stdio vs HTTP

Connecting it to LangGraph

관련 글

시스템은 언제나 알고, 로컬 효율과 시스템 성능은 별개의 문제

클로드 스킬에 포함할 4가지 문장

비전 LLMs도 PDF 파서… 차트·그래프 읽어 RAG 지원

앨라배마 진출 강화… 신규 투자·지역 사회 지원