2814 words
14 minutes
[Agent]AI Agent 디자인패턴: MCP와 A2A관련 패턴들
2026-01-05
2026-02-22
Machine Learning
agent

10. MCP(Model Context Protocol)#

Pattern Overview#

  1. Resources (리소스)
  • 외부 데이터 소스 (파일, DB, API 등)
  • URI로 식별
  • 읽기 전용
  1. Tools (도구)
  • 실행 가능한 기능
  • 입력/출력 스키마 정의
  • Agent가 호출
  1. Prompts (프롬프트)
  • 재사용 가능한 프롬프트 템플릿
  • 서버제공
  1. Sampling (샘플링)
  • 서버가 LLM 호출 요청
  • 클라이언트가 실행

Practical Applications & Usecases#

  1. Information Processing Workflows
  2. Complex Query Answering : sequential processing workflow
  3. Data Extraction and Transformation
  4. Content Generation Workflows
  5. Conversational Agents with State
  6. Code Generation and Refinement
  7. MultiModal and multi-step Reasoning

MCP 동작방식#

프로토콜로서의 MCP#

  • LLM자체는 다음에 올 토큰(단어)을 확률적으로 예측해서 생성하는 단하나의 기능을 가짐->즉, 기본적으로 텍스트 생성기임
  • LLM은 텍스트 생성만을 할 수있기 때문에 LLM이 외부도구를 호출하는 JSON text를 생성한 순간 생성을 멈추고 host머신한테 제어권을 넘긴 뒤 호스트머신의 도구 호출이 끝나면 다시 제어권을 받는 방식으로 외부도구와 상호작용함
  • 이를 Stop Sequence라 함
  • 이를 통해 LLM이 날씨 api호출 같은 외부도구를 호출하는 tool calling이 가능해짐

JSON RPC 2.0#

: MCP 클라이언트와 서버가 소통하는 명세

  • initialize로 handshake실행, tools/call로 클라이언트와 서버가 상호작용
  • REST api가 자원 중심이면 JSON RPC는 보다 행동 중심에 가깝다.
  • 언어·플랫폼 독립적인 RPC
  • REST처럼 리소스 모델링을 강요하지 않음
  • SOAP보다 가볍고 단순
  • gRPC처럼 바이너리/IDL 기반까지는 필요 없는 경우
{
  "jsonrpc": "2.0",
  "method": "subtract",
  "params": [42, 23],
  "id": 1
}

MCP Server#

import sys
import json


# --- 1. define tool
def add(a, b): return a + b
def subtract(a, b): return a - b


def main():
    # Stdio 방식으로 계속 입력을 대기함
    while True:
        try:
            # stdin
            line = sys.stdin.readline()
            if not line:
                break


            request = json.loads(line)
            method = request.get("method")
            msg_id = request.get("id")


            # --- 2. Handshake: 초기화 요청 처리 ---
            if method == "initialize":
                response = {
                    "jsonrpc": "2.0",
                    "id": msg_id,
                    "result": {
                        "protocolVersion": "2024-11-05",
                        "capabilities": {
                            "tools": {}  # 도구 기능 지원 선언
                        },
                        "serverInfo": {
                            "name": "simple-calculator",
                            "version": "1.0.0"
                        }
                    }
                }
                sys.stdout.write(json.dumps(response) + "\\n")
                sys.stdout.flush()


            # 3. check initialized
            elif method == "notifications/initialized":
                pass


            #4. show tools/list
            elif method == "tools/list":
                response = {
                    "jsonrpc": "2.0",
                    "id": msg_id,
                    "result": {
                        "tools": [
                            {
                                "name": "add",
                                "description": "두 숫자를 더합니다.",
                                "inputSchema": {
                                    "type": "object",
                                    "properties": {
                                        "a": {"type": "number"},
                                        "b": {"type": "number"}
                                    },
                                    "required": ["a", "b"]
                                }
                            },
                            {
                                "name": "subtract",
                                "description": "두 숫자를 뺍니다 (a - b).",
                                "inputSchema": {
                                    "type": "object",
                                    "properties": {
                                        "a": {"type": "number"},
                                        "b": {"type": "number"}
                                    },
                                    "required": ["a", "b"]
                                }
                            }
                        ]
                    }
                }
                sys.stdout.write(json.dumps(response) + "\\n")
                sys.stdout.flush()


            # --- 5. run actual function
            elif method == "tools/call":
                params = request["params"]
                tool_name = params["name"]
                args = params["arguments"]


                result_content = ""


                try:
                    if tool_name == "add":
                        val = add(args["a"], args["b"])
                        result_content = str(val)
                    elif tool_name == "subtract":
                        val = subtract(args["a"], args["b"])
                        result_content = str(val)
                    else:
                        raise ValueError("Unknown tool")
                except Exception as tool_err:
                    result_content = f"Error: {str(tool_err)}"


                response = {
                    "jsonrpc": "2.0",
                    "id": msg_id,
                    "result": {
                        "content": [
                            {
                                "type": "text",
                                "text": result_content
                            }
                        ]
                    }
                }
                sys.stdout.write(json.dumps(response) + "\\n")
                sys.stdout.flush()


        except Exception as e:
            sys.stderr.write(f"Server Error: {e}\\n")
            sys.stderr.flush()


if __name__ == "__main__":
    main()

Practical implementation#

  • fastmcp 예제
from fastmcp import FastMCP


# 1. MCP server instance
mcp = FastMCP("WeatherServer")


@mcp.tool()
def get_weather(city: str) -> str:
    # 실제로는 외부 API를 호출하는 로직이
    weather_db = {
        "seoul": "맑음, 25도",
        "tokyo": "비, 22도",
        "newyork": "흐림, 18도"
    }


    result = weather_db.get(city.lower(), "unknown")
    return f"[{city}]의 현재 날씨: {result}"


# 'resource://' 스키마를 통해 접근 가능한 데이터를 정의
@mcp.resource("resource://system/status")
def get_system_status() -> str:
    return "System OK - Server is running smoothly."


# client와 파이프로 통신
if __name__ == "__main__":
    mcp.run()

Concepts#

Concept

  • Model Context Protocol (MCP) : Anthropic이 제안한 AI-도구 연결 표준 프로토콜
  • MCP Client : Agent 측, 도구를 호출하는 주체
  • MCP Server : 도구/데이터를 제공하는 측
  • Stop Sequence : LLM이 도구 호출 텍스트(ex: JSON)를 완성하자마자 생성을 멈추게 하는 특정 문자열. 제어권을 코드(Host)로 가져오기 위해 필수적.
  • Raw Tool Calling : 표준 프로토콜 없이 프롬프트 엔지니어링과 문자열 파싱만으로 도구를 사용하는 원시적인 방식.
  • N * M Problem : 표준 없이 N개의 앱과 M개의 서비스를 개별 연결할 때 발생하는 비효율성.
  • capabilities : Handshake 단계에서 서버가 지원하는 기능(Tools, Resources, Prompts)을 명시하는 필수 필드.
  • Model-in-the-Loop : 도구 실행 과정에서 LLM이 ‘결정’을 내리고, 시스템이 ‘실행’한 뒤, 다시 LLM이 ‘해석’하는 순환 구조
  • Context Stuffing : 도구의 실행 결과를 다시 프롬프트에 끼워 넣어 LLM이 마치 방금 계산한 것처럼 알게 하는 기법

11. Goal Setting and Monitoring#

Agent에게 목적과 그 목적에 대한 tracking 수단을 부여하는 패턴

Pattern Overview#

  • 목적과 추적 능력: 이 패턴은 에이전트에게 단순한 작업 수행이 아닌 ‘목적(Purpose)‘을 부여하고, 진행 상황을 추적할 수 있는 메커니즘을 제공함
  • SMART 목표 설정: 목표는 Specific, Measurable, Achievable(달성 가능), Relevant(관련성 있음), Time-bound 의 기준을 따라야 함
  • 성공 기준과 지표의 정의: 효과적인 모니터링을 위해서는 성공을 판단할 수 있는 명확한 지표(Metrics)와 기준(Criteria)이 필수적입니다.
  • 모니터링: 모니터링은 단순히 에이전트의 행동만 보는 것이 아니라, 에이전트의 행동, 환경의 상태, 도구(Tool)의 출력값을 모두 관찰해야 함

Monitoring#

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, SystemMessage


# 1. State Definition
class AgentState(TypedDict):
    objective: str         # 최종 목표 (Goal)
    plan: List[str]        # 계획 단계
    current_step: int      # 현재 단계
    results: dict          # 실행 결과
    is_goal_met: bool      # 모니터링 결과 (성공 여부)


# 2. Nodes (각 단계의 로직)


def planner(state: AgentState):
    print(f"--- Planning for: {state['objective']} ---")
    # LLM을 사용해 계획 생성 로직 (생략)
    return {"plan": ["Search Info", "Draft Content", "Review"], "current_step": 0}


def executor(state: AgentState):
    step = state['plan'][state['current_step']]
    print(f"--- Executing: {step} ---")
    # 도구 실행 로직 (생략)
    return {"results": {step: "Done"}}


def monitor(state: AgentState):
    print("--- Monitoring Progress ---")
    # self reflection
    if state['current_step'] >= len(state['plan']) - 1:
        return {"is_goal_met": True}
    else:
        return {"is_goal_met": False, "current_step": state['current_step'] + 1}


# 3. Graph Construction (워크플로 연결)
workflow = StateGraph(AgentState)
workflow.add_node("planner", planner)
workflow.add_node("executor", executor)
workflow.add_node("monitor", monitor)
workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", "monitor")


# 4. Conditional Edge
def should_continue(state: AgentState):
    if state['is_goal_met']:
        return END  # 목표 달성 시 종료
    else:
        return "executor" # 달성 안될 시 재실행


workflow.add_conditional_edges(
    "monitor",
    should_continue
)


app = workflow.compile()


# 5. Execution
inputs = {"objective": "Write a report about AI Trends", "is_goal_met": False}
app.invoke(inputs)

Concept#

  • Goal Setting: 명확한 목표와 성공 기준을 정의하는 것
  • Goal Decomposition: 큰 목표를 작은 하위 목표로 분해
  • Progress Monitoring: 목표 달성 진행 상황을 추적
  • Goal Alignment Check: 현재 행동이 목표에 부합하는지 확인
  • Re-planning: 목표에서 벗어났을 때 경로 수정

12. Error Handling#

  • 통제 불가능한 현실 세계(API 중단, 네트워크 지연, 환각 등)에서 오류가 발생할 수 있음
  • 여기서는 통제불가능한 오류를 Agent가 어떻게 다루게 하는지에 대한 전략을 다룸

Pattern Overview#

1. 에러 감지#

  • 오류감지를 위한 트리거(API 오류, 타임아웃 등)
  • 출력검증 -> 도구의 출력스키마 검증

2. 대응 전략(Handling)#

  • Retries (재시도): 일시적인 네트워크 오류등일 경우 파라미터를 미세하게 조정하여 다시 시도
  • Fallbacks (대안 실행): fallback은 기준을 완화하는 것
  • Graceful Degradation : 전체 시스템을 멈추는 대신, 에러가 난 부분만 제외하고 나머지 기능을 수행

3단계: 복구 및 학습 (Recovery)#

  • State Rollback (상태 롤백): 에러 발생 이전 상태로 롤백 -> git worktree 참고
  • Self-Correction (자가 수정):
  • Escalation (이관): 에이전트가 해결 불가능하다고 판단하면 즉시 개발자에게 넘김

13. Human In the Loop#

AI가 처리하지 못하는 grey area를 인간이 보완하는 협업 아키텍처

Pattern Overview#

이관 (Escalation)#

: 인간에게 작업을 이관

승인 및 검토 (Approval & Review)#

: AI가 작업을 수행하기 직전(Before Action) 또는 수행 후(After Action) 인간의 승인을 받음

# workflow 정의 시 breakpoint설정
workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.add_node("human_review", human_review_node)


app = workflow.compile(interrupt_before=["human_review"])

능동적 수정 (Active Correction & Learning)#

인간이 AI의 오류를 실시간으로 수정하고, 이 데이터를 다시 강화학습에 사용하여 성능을 개선

14. Knowledge Retrieval#

  • LLM이 텍스트를 생성하기 전 외부 데이터스토어에서 관련 정보들을 검색하여 해당 정보를 활용해 더 최신이거나 특화된 답변을 제시하는 것
  • 기본적인 목적은 지식의 단절(고정된 데이터) 와 Hallucination을 해결하는 것

Embedding#

  • 문자열을 다차원 실수벡터로 변환하는 것

Vector Database#

  • 다차원 임베딩 벡터를 효율적으로 저장하고 빠르게 검색할 수있게 설계된 특수목적의 DB
  • RAG, Agent파이프라인에서 Long Term Memory 역할을 함
  • LLM의 Semantic Search(의도와 맥락을 파악하는 검색)을 위한 Context 정보 제공

Advanced RAG#

Agentic RAG#

  • 추론이 가능한 Agent가 검색과정을 주도
  • 문서의 최신여부를 검증 후 반영 (구글 검색)
  • 필요에 따라 다단계 추론 : 질문을 하위질문으로 쪼개서 검색
  • tradeoff : 비용과 복잡도가 증대됨

Graph RAC#

  • 노드와 엣지로 연결된 지식 그래프 활용
  • 파편화된 정보에서 연결고리 탐색 가능

벡터 DB 구축의 절차#

1. Load#

  • 텍스트,이미지, DB등에서 원시 데이터를 가져옴
  • 데이터 자체 검증을 위한 최소한의 전처리 수행

2. Split (Chunking)#

  • LLM은 기본 Context Window가 제한적이기에 데이터를 의미있는 Chunk로 쪼갤 필요가 있음

Splitting Strategies

  • Character Split: 단순 토큰 수 단위로 자르기.
  • Recursive Character Split : 문단 -> 문장 -> 단어 순으로, 의미가 깨지지 않도록 구분자(\n\n, \n, .)를 기준으로 자르기
  • Semantic Chunking: 의미가 변하는 지점을 AI가 판단하여 자르기.

3. Embedding#

  • 잘라낸 데이터 Chunk를 AI Embedding 모델을 통과시켜 다차원 벡터로 변환

4. Save#

  • 생성된 벡터를 DB에 저장
  • 다음의 3 요소가 저장됨
    • Vector Value: 임베딩 벡터
    • Original Content: 원본텍스트,이미지 경고
    • Metadata : 필터링 (검색) 을 위한 메타데이터
    • Index : 데이터 정렬 및 탐색을 위한 index

15. A2A Protocol#

  • AI Agent프레임워크 파편화 문제를 해결하기 위한 http기반 프로토콜
  • Agent Card를 통해 Agent들끼리 서로의 명세를 자동으로 파악하고 연결 가능

Pattern Overview#

기본적으로 A2A 개발은 다음의 4가지 절차로 구성됨

  1. Agent Card 작성: 내 에이전트의 스펙을 JSON으로 정의
  2. 서버 구축: HTTP 서버(FastAPI 등)를 띄우고 tasks/send 등의 표준 엔드포인트 개방
  3. 핸들러 연결: 요청이 들어오면 실제 로직(LangChain)을 실행할 핸들러를 연결
  4. 인증 처리: 헤더의 토큰을 검증하는 로직을 추가

Concept

  • Agent Card : json-rpc 형태로 정에 된 Agent의 정의와 스펙
  • Agent Discovery : .well-known/agent.json 과 같은 endpoint를 통해 다른 agent의 스펙을 확인하는 절차. A2A의 핵심

References#

[Agent]AI Agent 디자인패턴: MCP와 A2A관련 패턴들
https://yjinheon.netlify.app/posts/04ml/agent/agentic_design_pattern_03/
Author
Datamind
Published at
2026-01-05
License
CC BY-NC-SA 4.0
[Agent]AI Agent 디자인패턴: 리소스 핸들링과 가드레일 패턴들
[Agent]AI Agent 디자인패턴: 핵심 워크플로우 관련 패턴들
© 2026 Datamind. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Fuwari