MCP와 LangGraph 멀티 에이전트
실습 개요
이번 실습에서는 MCP (Model Context Protocol)와 LangGraph를 활용하여 멀티 에이전트 환경에서 동작하는 항공권 검색 에이전트를 구축해보겠습니다. MCP는 AI 모델이 외부 툴이나 서비스를 통합된 인터페이스로 사용할 수 있게 해주는 표준 프로토콜입니다. LangGraph는 이러한 에이전트의 워크플로를 그래프 형태로 구성하여 상태를 유지하고 제어할 수 있게 해주는 오케스트레이션 프레임워크입니다. 즉, MCP를 통해 외부 API 툴(예: 항공권 검색)을 제어하고, LangGraph를 통해 에이전트의 단계별 논리를 구조화합니다.
이 실습의 목표는 Amadeus 항공권 검색 API를 사용하는 MCP 서버(툴 제공자)와, 이를 호출하는 LangGraph 기반 항공권 검색 에이전트를 작성하여, 최종적으로 최저가 항공권 목록을 출력하는 것입니다. 실습을 통해 다음과 같은 내용을 배우게 됩니다:
- 프로젝트 폴더 구조와 주요 파일 역할 이해
- Miniconda 기반 Python 개발 환경 설정 및 필수 라이브러리 설치
- MCP 서버 코드 분석: FastMCP를 이용한 툴 정의와 서버 실행 방법
- VS Code MCP 설정 파일을 이용한 MCP 서버 통합 (GitHub Copilot 에이전트와의 연계 준비)
- LangGraph 멀티 에이전트 코드 분석: Planner/Shopper/Presenter 노드로 구성된 상태 그래프의 동작 원리
- 전체 프로젝트 실행 방법: 서버 기동, 에이전트 실행, 그리고 결과 확인
폴더 구조 설명
프로젝트의 폴더 구조와 각 구성 요소는 다음과 같습니다:
servers/폴더 – MCP 서버 코드를 포함합니다.amadeus_server.py: Amadeus API를 통해 항공권 정보를 조회하는 MCP 서버 구현 파일입니다.agents/폴더 – LangGraph 에이전트 코드를 포함합니다.flight_search_agent.py: 멀티 단계로 항공권을 검색하고 결과를 출력하는 에이전트 구현 파일입니다..env파일 – Amadeus API 인증에 필요한 클라이언트 ID/Secret 등의 환경변수를 저장합니다. (중요: API Key와 같은 민감 정보는 코드에 하드코딩하지 않고 .env에 보관).vscode/mcp.json파일 – VS Code에서 MCP 서버를 쉽게 관리하기 위한 설정 파일입니다. 워크스페이스에 MCP 서버를 등록하여 VS Code의 AI 에이전트 (예: Copilot Chat)에 툴을 제공할 수 있습니다.- 기타 구성 – Python 환경 설정 (Miniconda 가상환경 mcp_dev 등)과 필요 라이브러리(FastMCP, Amadeus SDK, LangGraph 등)가 있습니다.
위와 같은 구조를 통해, 서버(툴 제공)와 에이전트(툴 활용)의 코드가 분리되어 있으며, VS Code 설정을 통해 개발 편의성과 연동을 도모합니다.
Python 개발 환경 설정 (Miniconda)
실습을 진행하기 위해 Python 3.12 버전 기반의 가상환경을 구성합니다. 여기에서는 Miniconda를 사용하여 환경을 설정하고 필요한 패키지를 설치합니다:
Miniconda 환경 생성 및 활성화: 터미널에서 아래 명령을 실행해
mcp_dev라는 이름의 새 환경을 만들고 활성화합니다.conda create -n mcp_dev python=3.12 conda activate mcp_dev- 필수 라이브러리 설치:
mcp_dev환경에서 다음과 같은 Python 패키지를 설치해야 합니다.- FastMCP – MCP 서버 구축을 위한 프레임워크 (
pip install fastmcp) - Amadeus Python SDK – Amadeus 항공권 API 연동 라이브러리 (
pip install amadeus) - LangGraph – 에이전트 워크플로 그래프 구성 라이브러리 (
pip install langgraph) - LangChain MCP Adapters – LangGraph/LangChain 에이전트가 MCP 툴을 사용하도록 도와주는 어댑터 (
pip install langchain-mcp-adapters) - python-dotenv –
.env파일로부터 환경변수를 로드하기 위한 유틸 (pip install python-dotenv) - 기타: 예제 코드에서는
logging(표준 라이브러리) 등을 활용하므로 별도 설치가 필요 없습니다.
- FastMCP – MCP 서버 구축을 위한 프레임워크 (
VS Code 확장 및 Copilot: VS Code에서 GitHub Copilot Chat을 사용한다면, MCP 지원을 활성화해야 합니다. VS Code 1.102 이상에서는 MCP 서버와의 연동을 기본적으로 지원하므로, 사전에 Copilot Chat 확장과 MCP 허용 설정이 되어있는지 확인합니다 (설정에서 chat.mcp.access가 기본값이면 모든 MCP 서버 허용). 또한 이 프로젝트의 .vscode/mcp.json 설정을 이용하면 Copilot의 에이전트 모드에서 우리의 커스텀 MCP 툴을 사용할 수 있게 됩니다.
환경 변수 설정: 프로젝트 루트의
.env파일에AMADEUS_CLIENT_ID와AMADEUS_CLIENT_SECRET값을 할당해야 합니다. 이 값들은 Amadeus 개발자 사이트에서 발급받을 수 있습니다. (.env 파일 예시)AMADEUS_CLIENT_ID=YOUR_AMADEUS_API_KEY AMADEUS_CLIENT_SECRET=YOUR_AMADEUS_API_SECRET추가로, FLIGHT_AGENT_USE_SAMPLE 변수를 1로 설정하면 실제 Amadeus API를 호출하지 않고 샘플 데이터를 사용하도록 에이전트를 구성할 수 있습니다. API 크레덴셜이 없거나 호출을 제한하고 싶을 때 유용합니다.
이제 환경 구성이 완료되었으므로, 서버와 에이전트 코드를 살펴보겠습니다.
MCP 서버(servers/amadeus_server.py)
amadeus_server.py는 FastMCP 라이브러리를 사용하여 작성되었으며, Amadeus 항공권 검색 API를 하나의 MCP 툴(tool)로 노출합니다. 주요 내용을 단계별로 살펴보겠습니다.
FastMCP 서버 인스턴스 생성:
python mcp = FastMCP("flight_search")FastMCP클래스로 MCP 서버를 초기화합니다."flight_search"는 서버 이름(도메인)으로, 툴 식별 등에 사용됩니다. MCP는 클라이언트-서버 아키텍처를 따르며, 서버는 하나 이상의 툴을 제공하고 클라이언트는 이를 호출합니다.환경변수 로드 및 Amadeus Client 준비:
python ENV_PATH = PROJECT_ROOT / ".env" load_dotenv(dotenv_path=ENV_PATH, override=True) ... def _require_client() -> Client: cid = os.getenv("AMADEUS_CLIENT_ID") csec = os.getenv("AMADEUS_CLIENT_SECRET") if not cid or not csec: raise RuntimeError("AMADEUS_CLIENT_ID, AMADEUS_CLIENT_SECRET 환경변수 필요") return Client(client_id=cid, client_secret=csec).env파일의 환경변수를 로드한 후,_require_client함수에서 Amadeus API 클라이언트를 초기화합니다. 환경변수가 설정되지 않았을 경우 오류를 발생시켜 서버가 실행 중지되도록 합니다. Amadeus Python SDK의 Client 객체는 API 호출에 사용됩니다.- 항공권 검색 툴 정의:
@mcp.tool def amadeus_search(origin: str, destination: str, departureDate: str, returnDate: Optional[str] = None, adults: int = 1, ... ) -> Dict[str, Any]: """Amadeus 항공권 후보 조회 Tool.""" try: amadeus = _require_client() except Exception as e: return {"error": f"AuthError: {e}"} # API 파라미터 구성 params = { "originLocationCode": origin.strip().upper(), "destinationLocationCode": destination.strip().upper(), "departureDate": departureDate, "adults": adults, ... } if returnDate: params["returnDate"] = returnDate if travelClass: params["travelClass"] = travelClass ... # (기타 옵션 파라미터 설정) try: response = amadeus.shopping.flight_offers_search.get(**params) return {"offers": response.data} except ResponseError as e: # Amadeus SDK 오류 처리 if hasattr(e, "response") and hasattr(e.response, "result"): return {"error": e.response.result} return {"error": str(e)} except Exception as e: return {"error": f"UnhandledError: {e}"}@mcp.tool데코레이터를 사용하여 툴 함수를 정의합니다. 함수 이름amadeus_search가 바로 툴의 이름이 되며, 클라이언트는 이 이름으로 툴을 호출합니다. 이 함수는 Amadeus Flight Offers Search API에 쿼리를 보내 항공권 정보를 받아옵니다. 주요 동작을 살펴보면:- 매개변수:
origin,destination등 IATA 공항 코드와 날짜, 탑승객 수, 선호 좌석 등 다양한 옵션을 받을 수 있습니다. 실제 Amadeus API의 쿼리 파라미터와 대응되도록 구성했습니다. params딕셔너리: Amadeus API가 요구하는 키 (originLocationCode,departureDate등)에 맞게 전달인자를 매핑합니다. 불필요한 공백 제거나 대문자화 (upper())를 통해 입력을 정규화합니다.- Amadeus API 호출:
amadeus.shopping.flight_offers_search.get(...)메서드로 항공권 검색 API를 호출합니다. 이 API는 400여 개 항공사의 실시간 항공편 가격과 좌석 현황을 검색할 수 있는 강력한 API입니다. 결과 처리:
response.data에 검색된 항공권 리스트가 들어있습니다. 이를 그대로"offers"키 아래에 담아 반환합니다. 만약 API 호출에서 오류가 발생하면{"error": ...}형태로 오류 메시지를 내보냅니다. Amadeus SDK의ResponseError의 상세 내용(response.result)이 있을 경우 함께 반환하도록 했습니다.Note: MCP 툴 함수는 가능한 한 입력->출력의 순수한 함수 형태로 구현하는 것이 좋습니다. 상태를 가지지 않고, 입력 파라미터에 대응하는 결과(JSON 형태)를 돌려주면, MCP 프로토콜을 통해 이를 소비하는 측에서 결과 활용이 용이합니다.
- 매개변수:
로깅 및 예외 처리: 코드 상단에
logging설정이 있으며 DEBUG 모드 시 Amadeus SDK나 HTTP 통신의 잡다한 로그를 억제하는 구문이 있습니다. 이는 강의에서 상세히 다루진 않지만, 서버를 개발할 때 디버그용 로그와 불필요한 서드파티 로그를 조절하는 방법을 보여줍니다. 또한 툴 함수 내 try-except를 통해 인증 오류, API 오류, 기타 예외를 각각 처리하여 항상 일관된 JSON 형태 (offers또는error)로 결과를 반환하도록 했습니다.- 서버 실행:
if __name__ == "__main__": mcp.run(transport="http", host="127.0.0.1", port=8000, path="/mcp")
mcp.run()을 호출하여 MCP 서버를 기동합니다. 이 서버는 HTTP 프로토콜을 통해 동작하며, 127.0.0.1:8000 주소의 /mcp 엔드포인트를 listening 합니다. HTTP 스트리밍 전송 (transport="http")을 사용하므로, VS Code나 기타 MCP 클라이언트는 HTTP로 이 서버와 통신하게 됩니다. 이제 이 서버를 실행하면 "flight_search" MCP 서버가 뜨고, amadeus_search라는 툴을 외부로 노출하게 됩니다.요약하면, servers/amadeus_server.py는 Amadeus API 래퍼 MCP 서버입니다. 이 서버를 실행한 뒤에는, VS Code 또는 사용자 정의 에이전트(곧 작성할 LangGraph 에이전트)가 amadeus_search 툴을 호출하여 실제 항공권 검색 결과를 받아올 수 있습니다.
VS Code MCP 설정 (.vscode/mcp.json)
이제 작성한 MCP 서버를 VS Code에 통합해보겠습니다. .vscode/mcp.json 파일은 VS Code 워크스페이스에 MCP 서버를 등록하는 설정 파일입니다. VS Code에서 MCP 서버를 제어하고 Copilot 등의 AI 기능과 연동하기 위해 이 파일을 활용합니다.
mcp.json:
{
"servers": {
"flight_search": {
"type": "http",
"url": "http://127.0.0.1:8000/mcp"
}
}
}
VS Code에서 MCP 서버 사용하기:
위 설정을 저장하면 VS Code의 좌측 사이드바(확장 탭의 MCP 서버 목록)에 flight_search 서버가 나타납니다. 이를 마우스 오른쪽 클릭하거나 .vscode/mcp.json 파일 내 액션 lens를 통해 Start를 선택하면, VS Code가 해당 MCP 서버를 실행시킵니다. (주의: 먼저 python servers\amadeus_server.py 명령으로 서버를 기동해야 VS Code와의 연결이 설정됩니다.)
서버 기동 후, VS Code의 Copilot Chat 창을 열어 에이전트 모드(Agent Mode)로 전환하면, amadeus_search 툴이 사용 가능한 도구 목록에 나타납니다. 예를 들어, 채팅 프롬프트에 #amadeus_search라고 입력하면 해당 MCP 툴을 수동으로 호출할 수도 있고, 혹은 "ICN에서 JFK로 2025-10-01에 출발하는 가장 싼 항공편 알려줘" 같은 요청을 하면 Copilot이 이 툴을 적절히 호출하여 결과를 반환할 것입니다 (MCP 툴 자동 활용).
참고: VS Code의 MCP 통합은 GitHub Copilot Chat 활성화 시에만 의미가 있습니다. Copilot 없이도, 우리가 직접 작성한 LangGraph 에이전트로 MCP 서버를 호출해볼 수 있으므로, 다음 섹션에서는 코드 레벨에서 에이전트를 다루겠습니다.
LangGraph 멀티 에이전트(agents/flight_search_agent.py)
이제 MCP 서버를 활용하는 LangGraph 기반 항공권 검색 에이전트 코드를 분석해보겠습니다. 이 에이전트는 세 단계(노드)로 구성된 StateGraph를 정의하여, 사용자 입력을 처리하고 MCP 툴을 호출한 후 결과를 가공합니다.
전체 흐름 요약:
- Planner 노드 – 사용자의 질의나 입력을 정규화하여
search_params를 준비 - Shopper 노드 – MCP 툴 (
amadeus_search)을 호출하여 항공권offers리스트 획득 - Presenter 노드 –
offers에서 최저가 Top-10을 선별해 표 형태의result_table_pretty문자열 생성
각 부분을 상세히 살펴보겠습니다.
- 환경 및 클라이언트 초기화:LangChain MCP 어댑터의
MCP_CLIENT = MultiServerMCPClient({ "amadeus": { "url": "http://127.0.0.1:8000/mcp", "transport": "streamable_http", } })MultiServerMCPClient를 사용하여 MCP 서버 연결을 설정합니다."amadeus"라는 식별자에amadeus_server의 URL과 전송 방식을 등록했습니다. 이를 통해 코드 내에서MCP_CLIENT.get_tools()로 해당 서버의 툴 목록을 가져오고, 이후"amadeus_search"툴을 찾아 호출할 수 있게 됩니다. (만약 여러 MCP 서버를 쓴다면 이 딕셔너리에 추가하여 동시에 관리 가능) - 상태(State) 정의:
class FlightSearchState(TypedDict, total=False): user_input: Dict[str, Any] search_params: Dict[str, Any] offers: List[Dict[str, Any]] result_table: List[Dict[str, Any]] result_table_pretty: str error: strLangGraph의 상태는 TypedDict로 정의됩니다. 에이전트 실행 중에 공유할 데이터 키들을 명시한 것으로, 각 노드 함수가 이 상태를 읽고 쓰면서 결과를 누적합니다. 여기서는 사용자 입력, 검색 파라미터, 조회 결과 목록, 출력용 표 데이터, 예쁘게 꾸민 테이블 문자열, 에러 메시지 등을 상태의 키로 지정했습니다.
- Planner 노드 정의:Planner는 사용자 입력(
def make_planner_node(): def planner_node(state: FlightSearchState) -> FlightSearchState: print_phase("PLANNING", "START", "사용자 입력 정규화") user_input = state.get("user_input") if not user_input: # 기본값 처리 (미입력 시 경고 로그) ... user_input = { "origin": "ICN", "destination": "JFK", "departureDate": (date.today() + timedelta(days=30)).isoformat(), "adults": 1, } # 간단 정규화: origin, destination 대문자화, adults 정수 변환 for key in ("origin", "destination"): val = user_input.get(key) if isinstance(val, str): user_input[key] = val.strip().upper()[:3] if "adults" in user_input: try: user_input["adults"] = int(user_input["adults"]) except: user_input["adults"] = 1 new_state = dict(state) new_state["search_params"] = user_input print_phase("PLANNING", "DONE", f"origin={user_input.get('origin')} destination={user_input.get('destination')}") return new_state return planner_nodeuser_input)을 받아 검색에 적합한 형태로 가공합니다.주요 역할은:
- 필수 입력값이 없는 경우 기본값을 채워넣습니다. (예: 출발지/도착지 미지정 시 기본 “ICN→JFK”와 오늘부터 30일 후 날짜, 성인 1명으로 설정)
- 공항 코드는 대문자 3자리 IATA 코드로 변환하고, 성인 인원은 정수형으로 변환합니다.
- 가공된 입력을
search_params에 저장하여 다음 단계에서 사용할 수 있도록 합니다.
print_phase함수 호출로 현재 단계와 진행 상황을 콘솔에 출력합니다. 예를 들어[ 0.512s] [PLANNING] [DONE] origin=ICN destination=JFK와 같이 경과 시간과 함께 표시되어, 디버깅이나 흐름 추적에 도움을 줍니다. - Shopper 노드 정의: (비동기 함수로 정의)
def make_shopper_node(tools): async def shopper_node(state: FlightSearchState) -> FlightSearchState: print_phase("SHOPPING", "START", "Amadeus 항공권 조회") tool = tools.get("amadeus_search") if not tool: # 툴을 찾지 못한 경우 에러 처리 new_state = dict(state) new_state["offers"] = [] new_state["error"] = "amadeus_search tool not found" print_phase("SHOPPING", "ERROR", "Tool not found") return new_state params = state.get("search_params", {}) result = None async def _invoke_once(): try: if hasattr(tool, "ainvoke"): return await tool.ainvoke(params) return tool.invoke(params) except TypeError: # 인자전달 형태가 dict가 아닐 때 호환 if hasattr(tool, "ainvoke"): return await tool.ainvoke(**params) return tool.invoke(**params) def _extract_offers(res): # res에서 offers 리스트 추출하는 내부 함수... # 최대 5회 재시도 (스트리밍 응답 대기) offers = [] for attempt in range(5): try: result = await _invoke_once() except Exception as e: if attempt == 0: # 첫 호출에서 실패 시 바로 에러 처리 (fallback 결정) new_state = dict(state) new_state["error"] = str(e) if use_sample: new_state["offers"] = _sample_offers() else: new_state["offers"] = [] return new_state # 2회차 이상 실패는 무시하고 재시도 offers = _extract_offers(result) if offers: break await asyncio.sleep(1) # 잠시 대기 후 재시도 if not offers: # 최종 실패 처리 err_msg = None if isinstance(result, dict) and result.get("error"): err_msg = str(result.get("error")) new_state = dict(state) if err_msg: new_state["error"] = err_msg if use_sample: new_state["offers"] = _sample_offers() else: new_state["offers"] = [] print_phase("SHOPPING", "ERROR", new_state.get("error") or "no offers") return new_state new_state = dict(state) new_state["offers"] = offers print_phase("SHOPPING", "DONE", f"offers={len(offers)}") return new_state return shopper_nodeShopper는 실제로 항공권 검색 툴을 호출하는 역할을 합니다. 이 노드가 핵심적으로 MCP 서버와 통신하는 부분인데, 위 코드에서 특기할 점들을 정리하면:
tools매개변수:MCP_CLIENT.get_tools()를 통해 미리 받아온 툴 목록을 외부에서 클로저로 주입받습니다. 그런 다음 이름으로amadeus_search툴 객체를 찾습니다. 해당 툴이 없다면 (서버 미기동 등) 에러 상태를 기록하고 종료합니다.- MCP 툴 호출:
tool.invoke또는tool.ainvoke메소드를 사용합니다. FastMCP 툴은 일반적으로 비동기 스트리밍 응답을 지원하므로, 우선ainvoke(async invoke)을 시도하고, TypeError 발생 시 인자 전달 방식을 바꿔서 다시 시도합니다. (params딕셔너리째 전달 vs. 개별 인자 전달) - 스트리밍 및 재시도 로직: MCP 서버가 스트리밍 응답을 주는 경우, 첫 호출 시
result가 아직 부분 결과이거나offers가 비어있을 수 있습니다. 이를 대비해 최대 5회 반복해서 툴을 다시 호출해봅니다. 매 1초 간격으로 재시도하여, 응답 내offers필드에 데이터가 생기면 루프를 탈출합니다.- 실제 FastMCP의 HTTP 스트리밍 모드에서는 툴 실행 직후엔
{}또는{"offers": []}등 중간 결과가 올 수 있고, 최종적으로 스트림이 끝나면 complete 메시지와 함께 최종 JSON이 오게 됩니다. 그 과정을 코드에서 폴링(재시도) 방식으로 단순 처리한 것입니다.
- 실제 FastMCP의 HTTP 스트리밍 모드에서는 툴 실행 직후엔
- 결과 파싱:
_extract_offers(res)함수는 MCP 툴의 응답으로부터 항공권 리스트를 뽑아냅니다.res가 문자열이면 JSON 파싱을 하고, 딕셔너리이면res["offers"]나res["data"]등의 키를 점검합니다. Amadeus MCP 서버는{"offers": [...]}형태로 주지만, 혹시 모를 변형(예: LLM이 JSON을 텍스트로 감쌀 경우 등)을 고려해 폭넓게 탐색합니다. 이 함수는 응답 내에서 항공권 리스트를 찾는 로직으로, 상황에 따라 구조가 달라도 최대한 리스트를 얻도록 설계되었습니다. - 에러 처리와 폴백(fallback):
- 만약 첫 시도에 예외가 발생하면 (
except Exception as e), 곧바로 에러를 상태에 기록하고FLIGHT_AGENT_USE_SAMPLE환경변수가 켜져 있는 경우_sample_offers()라는 샘플 데이터를 대신 사용합니다. (샘플은 코드에 하드코딩된 하나의 예시 항공권입니다: ICN→JFK 직항, KE081편, 가격 1234.56) - 재시도를 다 했는데도
offers를 얻지 못한 경우도 마찬가지로, 오류 메시지가 있으면 기록하고 필요시 샘플 데이터를 채워넣습니다. 이렇게 하면 API 키 누락이나 일시적인 통신 문제 시에도 에이전트가 완전히 실패하지 않고, 학습이나 데모 목적의 샘플 결과를 보여줄 수 있게 됩니다.
- 만약 첫 시도에 예외가 발생하면 (
- 최종적으로
offers리스트 (항공편 옵션 목록)을 상태에 담고 다음 노드로 넘깁니다. 또한 완료 로그([SHOPPING][DONE] offers=X)를 출력합니다.
이 Shopper 노드 부분은 에이전트가 외부 도구(MCP 서버)와 상호작용하는 핵심으로, 실제 API 통신, 스트림 처리, 오류 대응까지 폭넓게 다루고 있습니다.
- Presenter 노드 정의: (비동기 함수, 출력 생성)
def make_presenter_node(): async def presenter_node(state: FlightSearchState) -> FlightSearchState: print_phase("PRESENTING", "START", "결과 표 생성") offers_list = state.get("offers", []) if state.get("error") and not state.get("offers"): # 에러 발생 시 안내 메시지 준비 msg = [ "오류로 인해 항공권을 불러오지 못했습니다.", f"서버 메시지: {state['error']}", "(FLIGHT_AGENT_USE_SAMPLE=1 설정 시 샘플 데이터 사용)" ] new_state = dict(state) new_state["result_table"] = [] new_state["result_table_pretty"] = "\n".join(msg) print_phase("PRESENTING", "ERROR", "error state") return new_state offers = offers_list if isinstance(offers_list, list) else [] def total_price(o): # 가격 추출 유틸 ... top = sorted(offers, key=total_price)[:10] header = ["순위", "편명", "출발지", "출발일시", "도착지", "도착일시", "등급", "운임"] lines = [" | ".join(header), " | ".join(["----"] * len(header))] def first_segment(o): ... def last_segment(o): ... for rank, offer in enumerate(top, 1): if not isinstance(offer, dict): continue fs = first_segment(offer); ls = last_segment(offer) flight_no = "-" extra_seg = 0 if fs: cc = fs.get("carrierCode") or ""; num = fs.get("number") or "" if cc or num: flight_no = f"{cc}{num}" # 경유편 개수 계산 (itineraries 내 segment 수 합산 - 1) its = offer.get("itineraries") if isinstance(its, list): for it in its: if isinstance(it, dict): segs = it.get("segments") if isinstance(segs, list) and len(segs) > 1: extra_seg += len(segs) - 1 if extra_seg: flight_no += f"(+{extra_seg})" dep_air = fs.get("departure", {}).get("iataCode") if fs else "-" dep_time = fs.get("departure", {}).get("at") if fs else "-" arr_air = ls.get("arrival", {}).get("iataCode") if ls else "-" arr_time = ls.get("arrival", {}).get("at") if ls else "-" cabin = "-" traveler_pricings = offer.get("travelerPricings") if isinstance(traveler_pricings, list) and traveler_pricings: tp0 = traveler_pricings[0] if isinstance(tp0, dict): fdbs = tp0.get("fareDetailsBySegment") if isinstance(fdbs, list) and fdbs: fd0 = fdbs[0] if isinstance(fd0, dict): cabin = fd0.get("cabin") or fd0.get("brandedFareLabel") or fd0.get("brandedFare") or "-" fare = "-" price = offer.get("price") if isinstance(price, dict): fare = price.get("grandTotal") or price.get("total") or price.get("amount") or "-" line = [str(rank), flight_no, dep_air, dep_time, arr_air, arr_time, cabin, fare] lines.append(" | ".join(line)) new_state = dict(state) new_state["result_table"] = top new_state["result_table_pretty"] = "\n".join(lines) print_phase("PRESENTING", "DONE", f"rows={len(top)}") return new_state return presenter_node
- **오류 상황 처리**: 만약 이전 단계까지 ```state["error"]```가 설정되어 있고 ```offers```가 비어있다면 (즉, API를 전혀 못 불러온 상황), 결과 대신 에러 메시지를 출력하도록 합니다. ```result_table_pretty```에 사용자에게 보여줄 안내 문자열을 담는데, 여기에는 서버 오류 메시지와 샘플 데이터 사용 팁이 포함됩니다. 이 상태로 바로 반환하면 에이전트 실행을 종료합니다.
- **정렬 및 상위 10개 추출**: ```offers``` 리스트를 ```total_price``` 기준으로 오름차순 정렬한 뒤 상위 10개 (가장 싼 10개)를 선택합니다. ```total_price``` 함수에서는 ```offer["price"]``` 객체 내에 ```"grandTotal"``` (총액)이나 ```"total"``` 등의 키를 찾아 실수(float)로 변환합니다. 가격 정보가 누락되었을 경우 아주 큰 값(```1e20```)을 줘서 정렬 시 뒤쪽으로 가도록 처리합니다.
- **표(Header) 구성**: Markdown 표 형태로 결과를 정리합니다. 우선 헤더 행과 구분자 행을 작성합니다. 컬럼은 **순위, 편명(항공사코드+편명), 출발지, 출발일시, 도착지, 도착일시, 등급(좌석 클래스), 운임**으로 구성하였습니다.
- **데이터 행 구성**: 각 항공권 (```offer```)에 대해:
- **편명 및 경유 표기**: 첫 구간(```first_segment```)의 항공사 코드(```carrierCode```)와 편명(```number```)을 결합하여 편명을 만듭니다. 예를 들어 대한항공(KE) 81편이면 ```KE81```로 표시합니다. 여러 구간(```segment```)이 있는 여정의 경우 ```(+)```로 추가 경유 편 수를 표시합니다. ```extra_seg``` 변수는 추가 경유 횟수를 계산합니다. 예를 들어 ```(+1)```이면 1회 경유(2개 구간)인 항공권입니다.
- **출발/도착 정보**: 첫 번째 구간의 출발 공항 코드와 출발 시각, 마지막 구간의 도착 공항 코드와 도착 시각을 추출합니다. 이렇게 하면 직항이든 경유든 전체 여정의 출발/도착지를 한눈에 볼 수 있습니다. 시간은 ```ISO8601``` 형태 (```YYYY-MM-DDThh:mm:ss```)로 제공되므로, 실제 출력 시에는 그대로 표시되지만, 필요에 따라 포맷을 다듬을 수도 있습니다.
- 좌석 등급: ```travelerPricings[0].fareDetailsBySegment[0].cabin``` 필드 등을 참조하여 좌석 클래스를 추출합니다. 일반적으로 ```ECONOMY```, ```BUSINESS``` 등으로 나오며, 일부 경우 ```brandedFareLabel``` 등의 정보가 있을 수 있어 보완적으로 확인합니다.
- **운임 (가격)**: ```offer["price"]``` 딕셔너리에서 ```grandTotal``` 또는 ```total``` 값을 가져옵니다. (Amadeus API의 응답에는 ```"grandTotal"```에 총액이 문자열로 제공됨) 통화는 API 호출 시 ```currencyCode```를 지정하지 않으면 기본 EUR 또는 USD 등으로 나올 수 있습니다. 본 코드에서는 통화를 특별히 표시하지 않고 숫자만 출력합니다.
- 모든 정보를 문자열로 변환하여 ```lines``` 리스트에 ' | '.join(...) 형태로 이어붙입니다. 이렇게 하면 Markdown의 테이블 포맷이 완성됩니다.
- **상태 업데이트 및 반환**: 완성된 ```lines```를 개행으로 합쳐 ```result_table_pretty```에 넣고, 상위 10개 오퍼를 ```result_table```에 별도로 리스트로 저장합니다. 이후 DONE 로그를 출력하고 새로운 상태를 반환합니다.
Presenter 노드를 거치면, 에이전트의 상태에는 사람이 보기 좋은 최종 결과 (result_table_pretty)가 준비됩니다.
- LangGraph StateGraph 구성 및 실행:
async def run_agent(user_input: Dict[str, Any]): tools = await get_tools() planner_fn = make_planner_node() shopper_fn = make_shopper_node(tools) presenter_fn = make_presenter_node() use_graph = os.getenv("FLIGHT_AGENT_USE_GRAPH", "0") in ("1", "true", "True") if use_graph: # LangGraph 그래프 방식 실행 graph = StateGraph(dict) graph.add_node("planner", planner_fn) graph.add_node("shopper", shopper_fn) graph.add_node("presenter", presenter_fn) graph.add_edge(START, "planner") graph.add_edge("planner", "shopper") graph.add_edge("shopper", "presenter") graph.set_entry_point("planner"); graph.set_finish_point("presenter") workflow = graph.compile() try: result = await workflow.ainvoke({"user_input": user_input}) except Exception as e: ... use_graph = False result = {} state = result if isinstance(result, dict) else {} if not use_graph: # 수동 파이프라인 실행 (LangGraph 실패 또는 비활성시) state = {"user_input": user_input} try: state = planner_fn(state) state = await shopper_fn(state) state = await presenter_fn(state) except Exception as e: print_phase("PIPELINE", "ERROR", str(e))핵심은
StateGraph로 노드와 엣지를 구성하여 에이전트의 실행 흐름을 정의하는 부분입니다.FLIGHT_AGENT_USE_GRAPH환경변수에 따라 LangGraph 사용 여부를 결정하게 해두었는데, 이는 그래프 실행에 문제가 생길 경우 수동 순차 실행으로 fallback하기 위함입니다. LangGraph 경로를 보면:StateGraph(dict)로 상태 그래프를 생성 (상태 타입은 dict 기반).- 세 개의 노드를 그래프에 추가하고,
START -> planner -> shopper -> presenter순으로 엣지를 연결합니다. - 시작 노드와 종료 노드를 설정한 뒤
graph.compile()로 그래프를 컴파일합니다. workflow.ainvoke(...)를 통해 비동기 실행하고, 최종 상태를 받아옵니다.- 만약 LangGraph 실행에 예외가 발생하면,
use_graph를 False로 바꾸고 예외를 로깅한 후 수동 경로로 넘어갑니다. - LangGraph가 disabled이거나 오류난 경우에는
planner_fn -> shopper_fn -> presenter_fn을 순차 호출하여 동일한 처리를 수행합니다. (비동기 호출 주의:shopper와presenter는async이므로await필요)
- 에이전트 실행 진입점:
if __name__ == "__main__": origin = input("출발지(공항코드, 예: ICN): ") or "ICN" destination = input("도착지(공항코드, 예: JFK): ") or "JFK" departureDate = input("출발일(YYYY-MM-DD, 예: 2025-10-01): ") or "2025-10-01" adults = input("성인 인원수(기본 1): ") or "1" try: adults = int(adults) except: adults = 1 user_input = { ... } asyncio.run(run_agent(user_input))
or "기본값" 처리가 되어 있습니다. 입력이 완료되면 user_input 딕셔너리를 만들고, run_agent 함수를 asyncio 이벤트 루프로 실행하여 앞서 정의한 Planner->Shopper->Presenter 일련의 과정을 거칩니다.에이전트 실행이 끝나면 ```result_table_pretty```를 콘솔에 출력하도록 되어 있습니다. 코드를 보면:
```python
print("\n=== 최종 Top-10 항공권 ===")
table = state.get("result_table_pretty")
if table:
print(table)
else:
... # fallback 출력 처리
if state.get("error"):
print(f"[오류] {state['error']}")
```
우선 결과 표가 있으면 그대로 출력하고, 없을 경우 (혹시 ```presenter``` 실패 등) 다시 한 번 최소한의 정보라도 출력하도록 예비 로직이 있습니다. 마지막으로 ```error```가 있다면 [오류] ... 형태로 별도로 표기합니다.
정리하면, agents/flight_search_agent.py는 MCP 서버의 툴을 호출하여 받은 항공권 데이터를 가공해주는 완결된 에이전트입니다. LangGraph를 통해 단계별로 모듈화했고, VS Code MCP 서버와 연계하여 동작하지만, 동시에 터미널 상에서 독립적으로 실행될 수도 있도록 하이브리드하게 작성되었습니다.
프로젝트 실행 방법
마지막으로, 지금까지의 코드를 실제로 실행하는 방법을 정리합니다:
- 환경 변수 설정:
.env파일에 Amadeus API의AMADEUS_CLIENT_ID와AMADEUS_CLIENT_SECRET값을 올바르게 넣었는지 확인합니다. (또는 API 키가 없다면FLIGHT_AGENT_USE_SAMPLE=1로 샘플 모드로 진행) - MCP 서버 실행: VS Code를 사용할 경우
.vscode/mcp.json에서 flight_search 서버를 Start합니다. 먼저 터미널에서 아래 명령으로 서버를 실행합니다:서버가 정상적으로 시작되면(mcp_dev) $ python servers/amadeus_server.pyhttp://127.0.0.1:8000/mcp에서 MCP 엔드포인트가 열리고, 콘솔 로그나 VS Code MCP 패널에amadeus_search툴이 등록되었음을 확인할 수 있습니다. - 에이전트 실행: 새로운 터미널을 열어 (서버 프로세스는 계속 동작 중이어야 함), 다음 명령을 실행합니다:그러면 콘솔에 차례로 출발지, 도착지, 출발일, 성인 인원수를 묻는 프롬프트가 나타납니다. 예시를 따라
(mcp_dev) $ python agents/flight_search_agent.pyICN,JFK,2025-10-01,1을 입력하거나 원하는 값으로 넣습니다 (또는 그냥 Enter로 기본값 사용). - 결과 확인: 입력이 완료되면 에이전트가 동작을 시작합니다.
- Planner 단계에서 입력을 정리하고
[PLANNING][DONE]로그를 출력합니다. - Shopper 단계에서 Amadeus API를 호출합니다. API 키가 유효하다면 실시간 항공권 정보를 받아오고
[SHOPPING][DONE] offers=N로그가 나옵니다 (N은 받아온 항공권 수, 기본 최대 20개). 만약 API를 호출할 수 없는 경우 샘플 데이터를 쓰고 오류 메시지를 남깁니다. - Presenter 단계에서 상위 10개 최저가 항공권을 선별해 표를 만듭니다.
[PRESENTING][DONE] rows=10로그와 함께 최종 결과를 출력합니다. - 콘솔에는 아래와 비슷한 테이블이 나타날 것입니다 (샘플 데이터 사용 시 예시):
=== 최종 Top-10 항공권 === 순위 | 편명 | 출발지 | 출발일시 | 도착지 | 도착일시 | 등급 | 운임 ---- | --------- | ------ | ------------------ | ------ | ------------------ | ------ | ---- 1 | KE81 | ICN | 2025-10-12T10:00:00 | JFK | 2025-10-12T15:00:00 | ECONOMY | 1234.56
위 출력은 예시로, 실제 Amadeus API 결과가 있다면 항공편 번호나 가격이 달라질 것입니다. 각 열의 의미는 순서대로 랭킹, 항공편 (경유 시 +n 표시), 출발 공항, 출발 시각, 도착 공항, 도착 시각, 좌석 등급, 총 운임입니다.
- Planner 단계에서 입력을 정리하고
- 에이전트와 MCP 통합 (선택): 만약 VS Code의 Copilot Chat과 연계해서 사용하고 있다면, 에이전트를 수동 실행하지 않고 Copilot에게 자연어로 질문할 수도 있습니다. 예를 들어, Copilot Chat 창에 에이전트 모드로 전환한 뒤 “ICN에서 JFK로 2025년 10월 1일에 출발하는 가장 저렴한 항공권은?“라고 물으면, Copilot이 자동으로
amadeus_searchMCP 툴을 호출하고 응답을 요약해줄 수 있습니다. 이때 우리의 MCP 서버가 반환한 raw JSON을 Copilot이 가공하여 답변하게 되는데, 만약 원하는 서식이 있다면 Presenter 노드에서 만든 것처럼 표 형태로 출력하도록 프롬프트를 튜닝할 수도 있습니다. (물론 본 실습에서는 독립 실행을 주로 다룹니다.)
모든 과정이 끝났으면, 필요에 따라 MCP 서버 프로세스를 종료합니다. VS Code MCP 패널에서 Stop을 누르고 서버 실행 터미널에서 Ctrl+C로 종료할 수 있습니다.
마무리: 전체 흐름 요약
이번 실습에서는 MCP와 LangGraph를 활용한 멀티 에이전트 항공권 검색기를 만들어보았습니다. 정리하자면 다음과 같습니다:
- MCP (Model Context Protocol): 외부 API 기능(항공권 검색)을 캡슐화하여 툴로 정의하고, 표준화된 프로토콜을 통해 에이전트가 이를 호출하도록 했습니다. 이를 통해 AI 어시스턴트(Copilot 등)가 손쉽게 실시간 기능을 활용할 수 있는 토대를 마련했습니다.
- LangGraph: 에이전트 내부 논리를 그래프 형태의 단계로 명확히 분리했습니다.
Planner/Shopper/Presenter로 역할을 구분함으로써 코드의 가독성과 유지보수성이 향상되었으며, LangGraph의 상태 관리와 흐름 제어를 경험해보았습니다. 이 접근은 복잡한 작업도 구조적으로 만들 수 있고, 병렬 처리나 조건 분기 등 확장에도 유리합니다. - VS Code 연동:
.vscode/mcp.json설정으로 VS Code IDE와 MCP 서버를 연결하여, 개발 편의성과 AI 도구(Copilot)와의 통합을 실현했습니다. 이를 통해 로컬에서 만든 MCP 서버를 바로 AI 에이전트에 활용하거나 채팅 답변에 반영할 수 있음을 확인했습니다. - 실행 및 결과: 실제 Amadeus API로부터 항공권 정보를 가져와 정렬/필터링하고, 최종적으로 사용자에게 유용한 형태의 정보(최저가 Top-10 표)를 제공하였습니다. API 인증이 없을 경우에도 샘플 데이터를 활용함으로써 개발 및 테스트를 진행할 수 있었습니다.
전체적으로, 외부 데이터 연동과 에이전트 로직을 분리하여 느슨한 결합 구조를 만든 것이 핵심입니다. MCP 서버는 하나의 서비스로서 항공권 데이터 제공에 집중하고, LangGraph 에이전트는 문제 해결 절차(계획->실행->응답)에 집중합니다. 이러한 설계는 실제 현업에서도 유용하며, 새로운 툴을 추가하거나 다른 API로 교체할 때 에이전트 부분을 크게 손대지 않고 확장할 수 있는 장점이 있습니다.
이번 강의를 통해 MCP와 LangGraph의 기본적인 사용법과 장점을 체험하셨기를 바랍니다. 더 나아가, 여러분은 이 구조를 응용하여 다양한 멀티에이전트 시나리오 (예: 호텔+항공 패키지 검색, 여행 일정 자동 구성 등)에도 도전해볼 수 있을 것입니다. 수고 많으셨습니다!