전체 활동내역

들어가며 이전에 살펴본 LangGraph로 웹 검색 Agent 만들기 (Web Search Agent Cookbook)에서는 LangGraph를 활용한 단일 에이전트 구축 방법을 다뤘다면, 이번에는 LangChain & LangGraph v1.0을 활용해 멀티 에이전트 시스템을 구축하는 방법을 알아보겠습니다. 기존 LangGraph에서 멀티 에이전트를 구축하려면 모델과 도구를 호출하는 노드를 일일이 정의하고, 워크플로우를 수동으로 연결해야 했습니다. 유연성은 높지만 시스템이 복잡해질수록 그래프 구조를 파악하기 어렵고 유지보수 부담이 커지는 문제가 있었죠. LangChain과 LangGraph가 v1.0으로 정식 출시되면서 AI 에이전트 개발이 한층 안정적이고 강력해졌습니다. 특히, LangChain v1.0부터 도입된 create_agent는 내부적으로 LangGraph 기반으로 구축되어 있어, LangGraph를 직접 다루지 않아도 Durable Execution, Streaming, Human-In-the-Loop 같은 강력한 기능들을 자동으로 활용할 수 있습니다. 또한, 미들웨어(Middleware) 기능으로 로깅, 프롬프트 수정, 에러 핸들링 같은 공통 로직을 깔끔하게 분리할 수 있게 되었습니다. 이번 쿡북에서는 LangChain v1.0과 HyperCLOVA X 모델을 활용해 Tool Calling 기반의 멀티 에이전트를 구축합니다. 웹 검색(Web Search), 글쓰기(Write), 저장(Save) 에이전트가 협업하여, 최신 정보를 검색하고, 목적에 맞는 글을 작성한 뒤, Notion이나 파일로 자동 저장하는 리포트 AI를 만들어봅니다. 이 가이드를 발판 삼아 여러분만의 창의적인 에이전트 시스템을 자유롭게 구축해 보세요. 사전 준비 멀티 에이전트 구축하기 프로젝트 진행을 위해서는 사전 준비 과정이 필요합니다. 각 과정마다 발급되는 키들을 환경 변수로 등록해 둡니다. API Key 발급 및 연동 설정 CLOVA Studio API CLOVA Studio 모델을 사용하기 위해 CLOVA Studio에서 API 키를 발급받아야 합니다. 본 예제에서는 HCX-005와 HCX-007 모델을 사용합니다. CLOVA Studio 접속 > 로그인 > 좌측 사이드바 'API 키' > 테스트 API 키 발급 발급된 키는 한 번만 표시되므로 반드시 복사하여 안전하게 보관하세요. API에 대한 자세한 내용은 CLOVA Studio API 가이드를 참고하시기 바랍니다. 네이버 검색 API 네이버 검색 오픈API를 활용하기 위해 네이버 개발자 센터에서 애플리케이션을 등록해야 합니다. 등록을 완료하면 Client ID와 Client Secret 정보를 확인할 수 있습니다. 자세한 내용은 네이버 개발자 센터의 애플리케이션 등록 가이드를 참고해 주세요. 네이버 개발자 센터 접속 > 로그인 > Application > 애플리케이션 등록 애플리케이션 등록 설정 애플리케이션 이름을 설정합니다 사용 API에서 '검색'을 선택합니다 비로그인 오픈 API 서비스 환경에서 'Web 환경'을 추가합니다. Tavily API 에이전트가 Tavily 웹 검색 기능을 사용하기 위해, Tavily API 키를 발급받아야 합니다. 무료 플랜의 경우 하루 1,000회까지 검색이 가능합니다. Tavily 웹 페이지 접속 > 로그인 > API Keys Notion API 에이전트가 개인 Notion 데이터베이스에 글을 저장하려면 Notion API Key와 업로드 대상 Data Source의 ID가 필요합니다. Notion API Key 발급 : Notion Developers 페이지 접속 > 우측 상단 'View my Integrations' > 로그인 > 새 API 통합 Data Source ID 확인 : 개인 Notion 접속 > API Key 발급시 설정한 워크스페이스로 이동 > 죄측 상단 '추가 옵션' > '데이터베이스' 선택 > 데이터베이스 설정 > 데이터 소스 관리 > 만들어진 데이터베이스 더보기 > 데이터 소스 ID 복사 프로젝트 구성 프로젝트의 전체 파일 구조는 다음과 같습니다. 프로젝트에 사용한 파이썬 버전은 3.11 입니다. multi-agent-cookbook/ ├── agent.py ├── utils/ │ ├── tool_agents.py │ ├── prompts.py │ ├── custom_middleware.py │ └── tools.py ├── langgraph.json ├── requirements.txt └── .env 환경 변수 설정 환경 변수를 설정합니다. 루트 디렉터리에 .env 파일을 생성한 뒤, 앞서 발급받은 API Key를 다음과 같이 입력하고 저장합니다. 이때 따옴표 없이 값을 작성해야 하며, VS Code에서 실행할 경우 설정에서 Use Env File 옵션이 활성화되어 있는지 확인하세요. CLOVA_STUDIO_API_KEY=nv-... OPENAI_API_KEY=sk-proj-... LANGSMITH_API_KEY=lsv2_pt_... LANGSMITH_TRACING=true LANGSMITH_PROJECT=Multi-agent-cookbook NAVER_CLIENT_ID=... NAVER_CLIENT_SECRET=... TAVILY_API_KEY=tvly-dev-... NOTION_API_KEY=ntn_... NOTION_DATA_SOURCE_ID=... 라이브러리 설치 프로젝트에 필요한 패키지 목록은 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 루트 디렉터리에 requirements.txt 파일로 저장하세요. requirements.txt 다운로드 루트 디렉터리에서 터미널을 실행하여 다음과 같이 예제 실행에 필요한 패키지를 설치합니다. 가상환경 설치를 권장합니다. # 1. 파이썬 가상환경 생성 python -m venv .venv # 2. 가상환경 활성화 (macOS/Linux) source .venv/bin/activate # (Windows) # .venv/Scripts/activate.ps1 # 3. 패키지 설치 pip install -r requirements.txt 멀티 에이전트 구축하기 멀티 에이전트 패턴 LangChain v1.0은 두 가지 멀티 에이전트 패턴을 제공합니다. 1. Tool Calling 패턴 중앙 Controller Agent가 모든 작업을 조율하고, 하위 에이전트들을 도구로 호출하는 구조입니다. 사용자와의 소통은 Controller Agent만 담당하며, Tool Agent는 Controller Agent의 도구로서 특정 작업만 수행합니다. Tool Calling 패턴으로 멀티 에이전트를 구현하면, 명확한 계층 구조와 모듈화로 관리가 쉽고 디버깅이 용이하지만, 모든 결정이 Controller를 거쳐야 하므로 토큰 소비량이 크고, 응답 시간이 다소 길어질 수 있습니다. 2. Handoffs 패턴 현재 에이전트가 다른 에이전트에게 제어권을 넘기는 방식으로, 각 에이전트가 순차적으로 사용자와 소통합니다. 이는 쿡북 작성일 기준으로 LangChain에서 아직 제공하지 않는 기능입니다. 이번 쿡북에서는 Tool Calling 패턴을 사용하며, 다음은 이를 기반으로 한 멀티 에이전트 아키텍처입니다. 관리자 에이전트(Controller Agent) 작업에 적합한 전문가 에이전트를 판단하여 호출하고, 그 결과를 받아 전체 흐름을 조율하는 오케스트레이터(Orchestrator) 역할을 수행합니다. LoggingMiddleware: 에이전트의 모든 실행 단계를 로깅하여 디버깅을 돕습니다. DynamicModelMiddleware: 대화 길이가 5개 보다 많아지면, 자동으로 더 강력한 모델로 전환합니다. HumanInTheLoopMiddleware: 민감한 도구 호출 전, 정지상태가 되는 interrupt를 발생시킵니다. 전문가 에이전트(Tool Agents) 웹 검색 에이전트(Web Search Agent): 네이버/Tavily API를 활용한 웹 검색을 수행합니다. 네이버 웹 검색 실패 시, 자동으로 Tavily 웹 검색으로 전환하여 웹 검색 안정성을 높입니다. 글쓰기 에이전트(Write Agent): 검색 결과를 바탕으로 리포트 또는 블로그 형식의 글을 작성합니다. 저장 에이전트(Save Agent): 작성된 콘텐츠를 Notion 또는 파일 시스템에 저장합니다. 미들웨어 구현하기 미들웨어는 에이전트의 실행 과정에 개입하여 기능을 확장하는 컴포넌트입니다. 미들웨어는 LangChain v1.0의 핵심 기능으로, 에이전트 로직을 수정하지 않고도 로깅, 프롬프트 변경, 에러 처리 등을 추가할 수 있게 합니다. 미들웨어가 개입할 수 있는 시점 미들웨어는 에이전트 실행의 다양한 시점에서 작동할 수 있습니다. Node-style hooks 특정 실행 지점에서 순차적으로 실행됩니다. 로깅, 유효성 검사, 상태 업데이트에 사용합니다. before_agent / after_agent: 에이전트 시작 전후 before_model / after_model: 모델 호출 전후 Wrap-style hooks 핸들러 호출을 가로채고 실행을 제어합니다. 재시도, 캐싱, 변환에 사용합니다. 핸들러를 0번(조기 종료), 1번(정상 흐름), 또는 여러 번(재시도 로직) 호출할지 결정할 수 있습니다. wrap_tool_call: 도구 실행 시 wrap_model_call: 모델 실행 시 Convenience dynamic_prompts: 동적 시스템 프롬프트 생성 미들웨어 구축 방식 LangChain에서는 목적과 복잡도에 따라 여러 방식으로 미들웨어를 구성할 수 있습니다. Decorator-based middleware: 단일 훅(Hook)만 필요한 간단한 로직을 적용하거나, 별도의 설정 없이 빠르게 프로토타이핑을 진행할 때 사용하면 좋습니다. Class-based middleware: 동기/비동기 처리를 모두 지원해야 하거나, 여러 개의 훅과 복잡한 설정을 하나의 모듈로 묶어 체계적으로 관리해야 할 때 적합합니다. Built-In Middleware: LangChain에서 제공하는 기본 미들웨어를 바로 사용할 수 있습니다. 이 프로젝트에서 사용할 미들웨어 이 프로젝트에서는 에이전트 실행 흐름을 제어하고 안정성을 높이기 위해 다음과 같은 미들웨어를 사용합니다. 커스텀 미들웨어를 구성할 때는 각 미들웨어가 실행 흐름에 개입하는 시점을 고려해야 하며, 이때 LoggingMiddleware는 미들웨어의 개입 지점을 이해하는 데 도움이 될 수 있습니다. HumanInTheLoopMiddleware : 민감한 도구 호출 전, 정지상태가 되는 interrupt를 발생시킵니다. LoggingMiddleware : 에이전트의 모든 실행 단계를 로깅하여 디버깅을 돕습니다. DynamicModelMiddleware : 대화 길이가 5개 보다 많아지면, 자동으로 더 강력한 모델로 전환합니다. NaverToTavilyFallbackMiddleware : 네이버 웹 검색 실패 시, 자동으로 Tavily 웹 검색으로 전환하여 웹 검색 안정성을 높입니다. writing_farmat : 사용자가 요청한 글 타입(report/blog)에 따라 동적으로 시스템 프롬프트를 변경합니다. # custom_middleware.py import os from typing import Any, Callable from langchain_naver import ChatClovaX from langchain.agents.middleware import AgentMiddleware, AgentState, ModelRequest, dynamic_prompt from langchain.agents.middleware.types import ModelResponse, ToolCallRequest from langchain.messages import ToolMessage from langgraph.types import Command from langgraph.runtime import Runtime from .tools import tavily_web_search from .prompts import WRITING_PROMPTS from dotenv import load_dotenv load_dotenv() class ReportAgentState(AgentState): content_type : str destination : str class LoggingMiddleware(AgentMiddleware): """에이전트 실행 과정 로깅""" async def abefore_agent(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None: print("\n" + "="*60) print(f"🔄 LoggingMiddleware.abefore_agent") print(f"🚀 에이전트 시작") print(f" 메시지 수: {len(state['messages'])}개") print("="*60) return None async def abefore_model(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None: print("\n" + "-"*60) print(f"🔄 LoggingMiddleware.abefore_model") if state['messages']: last = state['messages'][-1] print(f"🤖 모델 호출") print(f" 입력: {type(last).__name__}") if hasattr(last, 'content') and last.content: preview = last.content[:60] + "..." if len(last.content) > 60 else last.content print(f" 내용: {preview}") print("-"*60) print(f"\nController 응답 중..") return None async def aafter_model(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None: print("\n" + "-"*60) print(f"🔄 LoggingMiddleware.aafter_model") if state['messages']: last = state['messages'][-1] if hasattr(last, 'tool_calls') and last.tool_calls: tools = [tc['name'] for tc in last.tool_calls] print(f"🔧 도구 호출 예정") for tool in tools: print(f" → {tool}") elif hasattr(last, 'content') and last.content: print(f"✅ 모델 응답 완료") print("-"*60) return None async def aafter_agent(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None: print("\n" + "="*60) print(f"🔄 LoggingMiddleware.aafter_agent") print(f"🏁 에이전트 완료") print(f" 총 메시지: {len(state['messages'])}개") print("="*60 + "\n") return None async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: tool_name = request.tool_call.get('name', 'unknown') print("\n" + "-"*60) print(f"🔄 LoggingMiddleware.awrap_tool_call") print(f"⚙️ 도구 실행: {tool_name}") print("-"*60) print("\nTool Calling\n", end="", flush=True) result = await handler(request) print("\n" + "-"*60) print(f"🔄 LoggingMiddleware.awrap_tool_call (완료)") print(f"✅ 도구 완료: {tool_name}") if isinstance(result, ToolMessage) and result.content: preview = result.content[:100] + "..." if len(result.content) > 100 else result.content print(f" 결과: {preview}") print("-"*60) return result class DynamicModelMiddleware(AgentMiddleware): """대화 길이에 따라 모델 변경""" def awrap_model_call( self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse: CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") # 대화 길이에 따라 다른 모델 사용 model_name = "HCX-007" new_model = ChatClovaX(model=model_name, reasoning_effort="none", api_key=CLOVA_STUDIO_API_KEY) if len(request.messages) > 5: request.model = new_model print("\n" + "-"*60) print(f"🔄 DynamicModelMiddleware: Controller Using {model_name} for long conversation") print("-"*60) return handler(request) class NaverToTavilyFallbackMiddleware(AgentMiddleware): """네이버 검색 실패 시 Tavily로 폴백""" async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: tool_name = request.tool_call.get('name', 'unknown') if tool_name != 'naver_web_search': return await handler(request) try: result = await handler(request) # 네이버 검색 실행 print("\n" + "-"*60) print("🔄 NaverToTavilyFallbackMiddleware: ✅ 네이버 검색 성공") print("-"*60) return result except Exception as e: print("\n" + "-"*60) print(f"🔄 NaverToTavilyFallbackMiddleware: ⚠️ 네이버 검색 실패 → Tavily 검색 전환: {str(e)[:50]}...") print("-"*60) args = request.tool_call.get('args', {}) query = args.get('query', '') display = args.get('display', 5) # ainvoke 사용 (딕셔너리로 전달) tavily_result = await tavily_web_search.ainvoke({ "query": query, "display": display }) return ToolMessage( content=f"[Tavily 검색 결과]\n{tavily_result}", tool_call_id=request.tool_call.get('id', '') ) @dynamic_prompt def writing_format(request: ModelRequest) -> str: """사용자 요청에 알맞는 시스템 프롬프트 생성""" content_type = request.runtime.context.get("content_type", "report") base_prompt = "당신은 글쓰기 어시스턴트 입니다. 다음 형식으로 글을 작성하세요." print("\n" + "-"*60) print(f"🔄 writing_format: {content_type} 형식으로 작성") print("-"*60) if content_type == "blog": return f"{base_prompt}\n\n{WRITING_PROMPTS['blog']}" elif content_type == "report": return f"{base_prompt}\n\n{WRITING_PROMPTS['report']}" return base_prompt 관리자 에이전트(Controller Agent) 구성하기 관리자 에이전트는 사용자의 요청을 받아 적절한 전문가 에이전트를 호출하여 작업을 조율하는 중앙 관제 역할을 수행합니다. 이를 위해 HCX-007 모델을 사용하며, 구체적인 역할은 다음과 같습니다. 사용자 요청을 분석하고 필요한 전문가 에이전트를 선택합니다. 여러 전문가 에이전트를 순차적으로 호출하여 복합 작업을 수행합니다. Human-In-the-Loop을 통해 중요한 작업 전 사용자 승인을 받습니다. 최종 결과를 사용자에게 전달합니다. # agent.py import os import asyncio from langchain.agents import create_agent from langchain.messages import SystemMessage, HumanMessage from langchain.agents.middleware import HumanInTheLoopMiddleware from langchain_naver import ChatClovaX from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command from utils.tool_agents import call_write_agent, call_web_search_agent, call_save_agent from utils.custom_middleware import ReportAgentState, DynamicModelMiddleware, LoggingMiddleware from utils.prompts import CONTROLLER_PROMPT from dotenv import load_dotenv load_dotenv() # LangGraph Studio용 graph 생성 함수 def build_graph(): """LangGraph 서버에서 호출하는 graph 빌더""" CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") model = ChatClovaX(model="HCX-005", api_key=CLOVA_STUDIO_API_KEY) agent = create_agent( model=model, tools=[call_write_agent, call_web_search_agent, call_save_agent], checkpointer=InMemorySaver(), middleware=[ # LoggingMiddleware(), DynamicModelMiddleware(), HumanInTheLoopMiddleware( interrupt_on={ "call_web_search_agent": {"allowed_decisions": ["approve", "reject"]}, "call_write_agent": False, "call_save_agent": False } ) ], state_schema=ReportAgentState, system_prompt=CONTROLLER_PROMPT ) return agent async def main(): agent = build_graph() config = {"configurable": {"thread_id": "asd123"}} print("Multi Agent System Created!\n") while True: user_input = input("\nUser: ") if user_input.lower() in ["종료", "exit"]: print("AI: 대화를 종료합니다. 이용해주셔서 감사합니다.") break try: # 첫 실행 result = await agent.ainvoke( {"messages": [HumanMessage(user_input)]}, config=config ) # interrupt 확인 및 처리 while "__interrupt__" in result: print("\n" + "="*60) print("⏸️ 승인이 필요한 작업이 있습니다") print("="*60) interrupt_data = result["__interrupt__"][0].value action_requests = interrupt_data.get("action_requests", []) print(f"\n📋 총 {len(action_requests)}개의 작업 대기 중\n") decisions = [] for i, action in enumerate(action_requests, 1): tool_name = action.get("name", "unknown") tool_args = action.get("args", {}) print(f"작업 {i}:") print(f" 🔧 도구: {tool_name}") print(f" 📝 인자: {tool_args}") decision = input(f"\n\n승인하시겠습니까? (approve/reject): ").strip().lower() if decision == "approve": decisions.append({"type": "approve"}) print("✅ 승인됨\n") else: # approve가 아니면 모두 reject decisions.append({"type": "reject"}) print("❌ 거부됨\n") # 결정 전달 및 재실행 print("="*60) print("🔄 작업 재개 중...") print("="*60 + "\n") result = await agent.ainvoke( Command(resume={"decisions": decisions}), config ) # 최종 결과 출력 print("\nAI: ", end="", flush=True) final_message = result["messages"][-1] if hasattr(final_message, 'content'): print(final_message.content) else: print(final_message) print() except Exception as e: print(f"\n❌ 오류 발생: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main()) 전문가 에이전트(Tool Agents)구성하기 각 전문가 에이전트는 @tool 데코레이터로 래핑되어 관리자 에이전트가 호출할 수 있는 함수 형태가 됩니다. 작업 완료 후 Command 객체 반환을 통해 State를 업데이트하여 결과를 관리자 에이전트에게 전달합니다. 이번 프로젝트에서는 HCX-005 모델로 전문가 에이전트를 구성했습니다. 전문가 에이전트의 역할은 다음과 같습니다. @tool 데코레이터로 일반 함수를 LangChain 도구로 변환합니다. 각 Agent는 독립적인 create_agent로 생성되어 고유한 미들웨어와 도구를 가집니다. Command 객체를 통해 메시지와 상태를 업데이트하여 관리자 에이전트에 전달합니다. # tool_agents.py import os from typing import Annotated, Literal from langgraph.types import Command from langchain.tools import tool from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langchain.messages import HumanMessage, ToolMessage from langchain.tools import InjectedToolCallId from langchain_naver import ChatClovaX from .tools import naver_web_search, save_to_file, save_to_notion from .custom_middleware import writing_format, NaverToTavilyFallbackMiddleware from .prompts import WEB_SEARCH_PROMPT, SAVE_PROMPT from dotenv import load_dotenv load_dotenv() CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") write_model = ChatClovaX(model="HCX-005", api_key=CLOVA_STUDIO_API_KEY) web_search_model = ChatClovaX(model="HCX-005", api_key=CLOVA_STUDIO_API_KEY) save_model = ChatClovaX(model="HCX-DASH-002", api_key=CLOVA_STUDIO_API_KEY) write_agent = create_agent( model=write_model, middleware=[writing_format] ) web_search_agent = create_agent( model=web_search_model, tools=[naver_web_search], middleware=[NaverToTavilyFallbackMiddleware()], system_prompt=WEB_SEARCH_PROMPT ) save_agent = create_agent( model=save_model, tools=[save_to_notion, save_to_file], middleware=[HumanInTheLoopMiddleware( interrupt_on={ "save_to_notion":{"allowed_decisions": ["approve", "reject"]}, "save_to_file":{"allowed_decisions": ["approve", "reject"]} } )], system_prompt=SAVE_PROMPT ) @tool async def call_write_agent( article: str, content_type: Literal["report", "blog"], tool_call_id: Annotated[str, InjectedToolCallId], ) -> Command: """ 글쓰기 서브에이전트를 호출하여 지정된 형식의 글을 작성합니다. Args: article: 작성할 주제나 원본 내용 content_type: 글 형식 타입 ('report'|'blog') tool_call_id: LLM 도구 호출 ID (자동 주입) Returns: Command: 작성된 글과 업데이트된 상태를 포함하는 Command 객체 - messages: 작성된 글이 담긴 ToolMessage """ result = await write_agent.ainvoke( {"messages": [HumanMessage(article)]}, # context에 넣어서 middleware에 전달 context={ "content_type": content_type }, ) return Command(update={ "messages": [ ToolMessage( content=result["messages"][-1].content, tool_call_id=tool_call_id ) ], "content_type": content_type }) @tool async def call_web_search_agent( query: str, tool_call_id: Annotated[str, InjectedToolCallId], ) -> Command: """ 사용자가 검색을 요청하면 웹 검색 서브에이전트를 호출하여 웹 검색을 수행합니다 Args: query: 검색에 사용할 쿼리 runtime: 메인 에이전트의 상태 접근을 위한 런타임 객체 tool_call_id: LLM 도구 호출 ID (자동 주입) Returns: Command: 작성된 글과 업데이트된 상태를 포함하는 Command 객체 - messages: 작성된 글이 담긴 ToolMessage """ result = await web_search_agent.ainvoke( {"messages": [HumanMessage(query)]}, context={"current_agent": "web_search_agent"} ) return Command(update={ "messages": [ ToolMessage( content=result["messages"][-1].content, tool_call_id=tool_call_id ) ] }) @tool async def call_save_agent( content: str, destination: Literal["file", "notion"], filename_or_title: str, tool_call_id: Annotated[str, InjectedToolCallId], ) -> Command: """ 콘텐츠를 파일 또는 Notion에 저장합니다. Args: content: 저장할 콘텐츠 destination: 저장 위치 ('file' 또는 'notion') filename_or_title: 파일명(file) 또는 노션 페이지 제목(notion) tool_call_id: LLM 도구 호출 ID (자동 주입) Returns: Command: 저장 결과를 포함하는 Command 객체 """ # destination에 따라 다른 메시지 전달 if destination == "notion": context_msg = f"다음 내용을 Notion에 '{filename_or_title}' 제목으로 저장해주세요:\n\n{content}" else: # file context_msg = f"다음 내용을 '{filename_or_title}' 파일로 저장해주세요:\n\n{content}" result = await save_agent.ainvoke( {"messages": [HumanMessage(context_msg)]}) return Command(update={ "messages": [ ToolMessage( content=result["messages"][-1].content, tool_call_id=tool_call_id ) ] }) 도구 구현하기 전문가 에이전트인 웹 검색 에이전트와 저장 에이전트가 사용하는 도구를 구현합니다. 이 도구들은 멀티 에이전트와 미들웨어의 활용 방법을 보여주기 위한 예시로 구성되었습니다. 웹 검색 에이전트의 경우 naver_web_search 도구를 가지고 있고, NaverToTavilyFallbackMiddleware에 의해 tavily_web_search 도구 또한 활용할 수 있습니다. 저장 에이전트는 개인 노션에 업로드 할 수 있는 save_to_notion, 시스템에 저장할 수 있는 save_to_file 두 가지 도구를 가지고 있습니다. # tools.py import os import re import httpx from pathlib import Path from langchain.tools import tool from tavily import TavilyClient from notion_client import Client from dotenv import load_dotenv # 환경 변수 로드 load_dotenv() NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID") NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET") TAVILY_API_KEY = os.getenv("TAVILY_API_KEY") @tool async def naver_web_search(query: str, display: int = 10) -> dict: """ 네이버 검색 API를 호출해 결과를 구조화하여 반환합니다. Args: query: 검색어 display: 검색 결과 수(1~100) Returns: { "query": str, "total": int, "items": [{"title": str, "link": str, "description": str}]} } """ url = "https://openapi.naver.com/v1/search/webkr.json" params = {"query": query, "display": display} headers = { "X-Naver-Client-Id": NAVER_CLIENT_ID, "X-Naver-Client-Secret": NAVER_CLIENT_SECRET, } async with httpx.AsyncClient() as client: r = await client.get(url, headers=headers, params=params) r.raise_for_status() data = r.json() results = [] for item in data.get("items", []): results.append({ "title": re.sub(r"<.*?>", "", item.get("title") or "").strip(), "link": item.get("link"), "description": re.sub(r"<.*?>", "", item.get("description") or "").strip(), }) return { "query": query, "total": data.get("total", 0), "items": results, } # 본 쿡북에서는 해당 도구를 NaverToTavilyFallbackMiddleware에서 풀백 도구로 활용하였습니다 @tool async def tavily_web_search(query: str, display: int = 5) -> str: """ Tavily API를 사용하여 웹 검색을 수행합니다. Args: query: 검색어 display: 검색 결과 수(1~100) """ try: tavily_client = TavilyClient(api_key=TAVILY_API_KEY) # Tavily는 동기 라이브러리이므로, 비동기 처리를 위해 asyncio.to_thread 사용 import asyncio response = await asyncio.to_thread( tavily_client.search, query=query, max_results=display ) # 결과 포맷팅 results = [] for idx, result in enumerate(response.get('results', []), 1): result_text = f"\n{idx}. {result.get('title', 'No title')}\n" result_text += f" URL: {result.get('url', 'No URL')}\n" result_text += f" 내용: {result.get('content', 'No content')}\n" results.append(result_text) if not results: return "검색 결과가 없습니다." return "".join(results) except Exception as e: return f"검색 중 오류 발생: {str(e)}" @tool def save_to_file(filename: str, content: str) -> str: """ 지정된 파일 이름으로 리포트 내용을 시스템에 마크다운 파일로 저장합니다. Args: filename: 저장할 파일 이름 (확장자 없으면 자동으로 .md 추가) content: 파일에 저장할 텍스트 내용 Returns: 저장 성공/실패 메시지와 파일 경로 """ try: # 확장자 없으면 자동으로 .md 추가 if not filename.endswith(('.md', '.markdown', '.txt')): filename = f"{filename}.md" # 저장 디렉토리 생성 (없으면) save_dir = Path("reports") # 또는 원하는 디렉토리 save_dir.mkdir(exist_ok=True) # 전체 경로 file_path = save_dir / filename # 마크다운 형식으로 저장 with open(file_path, "w", encoding="utf-8") as f: f.write(content) # 절대 경로 반환 abs_path = file_path.resolve() return f"✅ 마크다운 파일로 저장 완료!\n📁 경로: {abs_path}" except Exception as e: return f"❌ 파일 저장 중 오류 발생: {e}" @tool def save_to_notion(page_title: str, content: str) -> str: """ Notion 데이터베이스에 새 페이지를 생성하고 콘텐츠를 저장합니다. 사용자가 작성한 글, 보고서, 메모 등을 Notion에 저장할 때 사용합니다. 페이지 제목과 본문 내용을 받아서 지정된 Notion 데이터베이스에 자동으로 추가합니다. Args: page_title: Notion 페이지의 제목 (예: "주간 보고서", "회의 내용") content: 페이지 본문에 저장할 텍스트 내용 (마크다운 형식 지원) Returns: 성공 시: 생성된 페이지 제목과 URL 실패 시: 오류 메시지 Examples: - "이 리포트를 Notion에 '월간 분석'이라는 제목으로 저장해줘" - "방금 작성한 글을 Notion에 저장" """ try: notion = Client( auth=os.getenv("NOTION_API_KEY"), notion_version="2025-09-03" # 최신 버전 ) data_source_id = os.getenv("NOTION_DATA_SOURCE_ID") # data_source_id 사용 new_page = notion.pages.create( parent={ "type": "data_source_id", "data_source_id": data_source_id }, properties={ "title": { # 기본 title "title": [ {"text": {"content": page_title}} ] } }, children=[ { "object": "block", "type": "paragraph", "paragraph": { "rich_text": [ {"text": {"content": content}} ] } } ] ) return f"✅ '{page_title}' 생성 완료: {new_page['url']}" except Exception as e: return f"❌ 오류: {str(e)}" 프롬프트 프로젝트에서 사용하는 모든 프롬프트를 정의합니다. 각 에이전트의 역할과 동작 방식을 명확히 지정하여 일관된 응답을 보장합니다. 사용된 프롬프트는 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 prompt.py 파일로 저장하세요. prompts.txt 다운로드 실행 및 동작 예시 다음은 HumanInTheLoopMiddleware가 적용된 멀티 에이전트 시스템의 동작 흐름입니다. 각 에이전트의 도구 호출 과정에서 사용자 승인 단계가 개입되며, 전체 응답 흐름은 다음과 같습니다. 에이전트 실행 결과는 다음과 같습니다. User: 이번 네이버 DAN25에서 발표된 네이버의 AI 전략에 대해 찾아보고 리포트로 작성해서 노션에 업로드 해줘 ============================================================ ⏸️ 승인이 필요한 작업이 있습니다 ============================================================ 📋 총 1개의 작업 대기 중 작업 1: 🔧 도구: call_web_search_agent 📝 인자: {'query': '네이버 DAN25에서 발표된 네이버의 AI 전략'} 승인하시겠습니까? (approve/reject): approve ✅ 승인됨 ============================================================ 🔄 작업 재개 중... ============================================================ ------------------------------------------------------------ 🔄 NaverToTavilyFallbackMiddleware: ✅ 네이버 검색 성공 ------------------------------------------------------------ ------------------------------------------------------------ 🔄 writing_format: report 형식으로 작성 ------------------------------------------------------------ ============================================================ ⏸️ 승인이 필요한 작업이 있습니다 ============================================================ 📋 총 1개의 작업 대기 중 작업 1: 🔧 도구: save_to_notion 📝 인자: {'page_title': '네이버 AI 전략 DAN25 발표 분석', 'content': '# 네이버의 AI 전략: DAN25 발표 내용 분석\n\n## 1. 개요\n본 리포트는 네이버의 DAN25 발표 내용을 바탕으로 한 AI 전략에 대해 분석합니다. 네이버는 AI 에이전트 도입 확대, 핵심 제조 산업 경쟁력 강화, 새로운 AI 도구 및 플랫폼 전략 공개, AI 산업 거품론 대응, 그리고 미래 비전과 글로벌 확장 계획을 통해 AI 기술의 발전과 실질적 가치 창출을 목표로 하고 있습니다.\n\n## 2. 핵심 발견사항\n- **AI 에이전트 도입 확대**: 네이버는 주요 서비스에 AI 에이전트를 도입해 개인화된 사용자 경험 제공.\n- **핵심 제조 산업 경쟁력 강화**: 반도체, 자동차, 조선 등 제조 산업에서의 AI 활용 방안 모색.\n- **신규 AI 도구와 플랫폼 전략**: 산업별 버티컬 AI와 경량화 모델을 통한 실질적인 가치 창출 목표.\n- **산업 거품론 대응**: 경량화 모델과 산업 특화 AI를 통해 실질적인 가치 창출 중요성 강조.\n- **글로벌 확장 계획**: 차세대 AI 전략 발표 및 글로벌 시장 진출 도모.\n\n## 3. 분석 및 인사이트\n네이버는 AI 기술의 전방위적 도입을 통해 사용자 맞춤형 서비스 제공을 강화하고 있으며, 제조업 분야에서도 AI 트랜스포메이션을 추진 중입니다. 또한, DAN25에서는 산업별 맞춤형 AI 솔루션과 경량화된 모델을 통해 실질적인 성과를 도출하고자 했습니다. AI 산업 내 거품론을 경계하며 실체 있는 기술 개발에 주력하고 있고, 글로벌 확장을 위한 미래 비전을 제시하고 있습니다.\n\n## 4. 결론 및 제언\n네이버의 AI 전략은 다각도로 전개되고 있으며, 이는 궁극적으로 사용자 경험 개선과 산업 전반의 혁신을 촉진할 것입니다. 향후 네이버는 AI 기술의 고도화와 더불어 글로벌 시장에서의 입지 강화를 위해 지속적인 투자와 연구 개발이 필요합니다.\n\n## 5. 참고 자료\n[AI 에이전트 도입 확대](https://www.etnews.com/20251023000297), [핵심 제조 산업 경쟁력 강화](http://www.efnews.co.kr/news/articleView.html?idxno=124617), [신규 AI 도구와 플랫폼 전략](https://www.asiatoday.co.kr/kn/view.php?key=20250930010016279), [AI 산업 거품론 대응](https://www.econovill.com/news/articleView.html?idxno=717550), [미래 비전과 글로벌 확장 계획](https://www.kmjournal.net/news/articleView.html?idxno=4023)'} 승인하시겠습니까? (approve/reject): approve ✅ 승인됨 ============================================================ 🔄 작업 재개 중... ============================================================ ------------------------------------------------------------ 🔄 DynamicModelMiddleware: Controller Using HCX-007 for long conversation ------------------------------------------------------------ AI: 리포트가 성공적으로 노션에 저장되었습니다! ### 세부 사항 - **제목:** 네이버 AI 전략 DAN25 발표 분석 - **노션 링크:** [네이버 AI 전략 DAN25 발표 분석](https://www.notion.so/AI-DAN25-2bf87d6d35378107a508c5b0bc8f478a) 추가로 필요하신 것이 있으면 말씀해 주세요! LangSmith Studio LangSmith Studio는 AI 에이전트 개발을 위한 전용 IDE(통합 개발 환경)입니다. 그래프 기반 시각화 인터페이스로 에이전트가 실행되는 동안 각 노드의 전환과 상태 변화를 실시간으로 추적할 수 있어, 복잡한 로직의 흐름을 한눈에 파악할 수 있습니다. 또한 프롬프트를 수정하면 즉시 반영되는 hot-reloading 기능으로 빠른 반복 개발이 가능하고, 멀티턴 대화를 테스트할 수 있는 Chat UI가 내장되어 있습니다. LangSmith API 에이전트의 동작 과정을 모니터링하고, LangSmith Studio를 사용하기 위해 LangSmith의 API 키를 발급받아야 합니다. LangSmith 접속 > 로그인 > 좌측 사이드바 'Settings' > API Keys 탭 > + API key langgraph.json LangGraph 애플리케이션의 구성 정보를 담고 있는 설정 파일입니다. 그래프의 위치, 의존성, 환경 변수 등 에이전트의 구조를 정의합니다. 자세한 설정은 공식 문서를 참고하세요. { "dependencies": ["."], "graphs": { "controller": "./agent.py:build_graph" }, "env": ".env" } LangSmith Studio 활성화 다음 명령어를 통해 LangSmith Studio를 활성화합니다. 다음 명령어를 실행하면 다음 과정이 순차적으로 수행됩니다. langgraph dev langgraph-cli가 langgraph.json 파일을 읽습니다. 로컬 API 서버가 시작됩니다. LangSmith Studio의 웹 UI가 이 서버에 자동으로 연결됩니다. 다음 URL을 통해 LangSmith Studio에 접근할 수 있습니다. https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024 LangSmith Studio에서는 내장된 Chat UI로 직접 구축한 에이전트와 대화하며 실시간으로 테스트할 수 있고, 토큰 소모량, 실행 시간, 각 컴포넌트의 입출력 등을 모니터링하여 디버깅과 최적화를 진행할 수 있습니다. 또한, 프롬프트를 수정하면 즉시 반영되어 다양한 프롬프트 변형을 빠르게 실험할 수 있으며, 그래프 시각화를 통해 에이전트의 실행 흐름을 직관적으로 파악할 수 있습니다. 마무리 이번 쿡북을 통해 LangChain v1.0이 제시하는 새로운 멀티 에이전트 구축 방법을 익혔습니다. 특히, create_agent와 미들웨어를 적극 활용함으로써 로깅, 에러 처리, Human-In-the-Loop 같은 핵심 기능을 모듈화할 수 있었고, 복잡한 그래프 구축 과정 없이도 유지보수가 용이한 에이전트 설계가 가능해졌습니다. 또한, HyperCLOVA X를 기반으로 관리자 에이전트가 하위 전문가 에이전트들을 정교하게 조율하는 협업 구조를 구현할 수 있었습니다. 이제 이 가이드를 발판 삼아, LangChain v1.0의 유연한 구조 위에 CLOVA Studio의 강력한 모델들을 더해 여러분만의 창의적인 에이전트 서비스를 완성해보시기 바랍니다.

안녕하세요. 정확한 확인을 위해 확인 가능한 url 알려주시면 디버깅해보겠습니다.

LLM 기반 에이전트를 만들고 나면, 그다음 고민은 얼마나 잘 작동하느냐입니다. 초기 프롬프트가 단순한 데모 상황에서는 만족스러운 답변을 내놓더라도, 실제 서비스 환경에서는 응답 품질이 떨어지거나, 의도와 다른 행동을 보이거나, 특정 입력에 취약한 패턴이 드러날 수 있습니다. 이럴 때는 데이터를 보강하고, 프롬프트를 다듬고, 정책을 조정해 에이전트를 점진적으로 더 똑똑하고 안정적으로 만드는 과정이 필수적입니다. 이번 쿡북에서는 이러한 개선 과정을 손쉽게 반복 실행할 수 있도록 도와주는 프레임워크, Agent Lightning을 다룹니다. 특히 별도의 모델 튜닝 없이도 프롬프트를 자동으로 수정·검증해 주는 APO(Automatic Prompt Optimization)를 활용해, 최소한의 설정만으로 에이전트 개선 루프를 구성하는 방법을 소개합니다. 또한 기본적으로 영문 프롬프트 최적화에 맞춰 설계된 Agent Lightning의 APO를 한국어 환경에서도 안정적으로 활용할 수 있도록, POML(Prompt Optimization Markup Language) 템플릿을 한국어 기반으로 커스터마이징해 적용하는 방법도 함께 다룹니다. 이번 쿡북을 통해 CLOVA Studio에서 제공하는 모델을 더 안정적으로 다루고, 실제 서비스 품질을 높이는 프롬프트 개선 전략을 익히는 데 도움이 되길 바랍니다. 1. Agent Lightning 개요 마이크로소프트에서 공개한 Agent Lightning은 에이전트의 학습과 최적화를 체계적으로 수행할 수 있도록 설계된 프레임워크입니다. 이 프레임워크는 에이전트의 실행을 자동으로 추적하고, 그 결과로 얻은 보상(Reward)을 기반으로 프롬프트나 정책을 개선할 수 있게 해줍니다. 1-1. 핵심 개념 Agent Lightning에서 다루는 핵심 개념은 다음과 같습니다. Task(태스크): 에이전트에게 주어지는 구체적인 입력 또는 임무입니다. 장소를 예약하거나 수학 문제를 풀어주는 것처럼, 에이전트가 해결해야 할 대상을 의미합니다. Rollout(롤아웃): 하나의 태스크가 주어지고, 에이전트가 실행되어 도구 호출이나 LLM 호출 등을 거쳐 행동을 완료하고, 마지막에 보상(Reward) 을 받는 한 번의 전체 사이클을 말합니다. Span(스팬): 롤아웃 내부의 작은 단위 실행입니다. LLM 호출 하나, 툴 실행 하나 등이 각각의 스팬이 될 수 있습니다. Prompt Template(프롬프트 템플릿): 태스크를 해결하기 위해 에이전트가 사용하는 지시문 및 프롬프트의 구조입니다. 이 템플릿은 알고리즘에 의해 반복적으로 개선됩니다. 에이전트가 수행하는 모든 롤아웃은 보상 정보와 함께 기록되고, 이 데이터를 기반으로 프롬프트나 정책을 점진적으로 개선할 수 있습니다. 1-2. 구성 요소 Agent Lightning은 다음 세 가지 주요 구성 요소로 이루어집니다. Agent(에이전트): 태스크를 입력받아 에이전트 로직을 수행하고 보상을 리턴합니다. 이를 통해 각 실행이 자동으로 롤아웃으로 기록됩니다. Algorithm(알고리즘): 프롬프트나 정책을 개선하기 위한 알고리즘입니다. APO, VERL 등 다양한 알고리즘을 지원하며, 이번 쿡북에서는 프롬프트 최적화를 다루기 위해 APO를 사용합니다. Trainer(트레이너): 에이전트와 알고리즘을 연결하고, 학습 루프를 제어하는 구성 요소입니다. 반복적인 실행과 평가를 통해 점진적인 개선을 수행합니다. 즉, 이미 만들어둔 에이전트 코드에 간단한 데코레이터(@rollout)를 추가하기만 하면, 각 실행의 입력, 출력, 보상 데이터를 자동으로 기록하고, 이를 기반으로 프롬프트나 정책을 개선하는 학습 가능한 에이전트 루프를 구성할 수 있습니다. 이러한 Agent Lightning을 활용하면 복잡한 학습 코드를 직접 작성하지 않아도, 에이전트의 실행 기록과 보상 정보를 바탕으로 다양한 프롬프트 버전을 자동으로 생성·평가할 수 있습니다. 그 과정에서 더 높은 보상을 주는 프롬프트가 자동으로 선택되고, 테스트셋 기준의 성능 비교와 기록까지 이루어져, 에이전트를 점진적으로 고도화하는 작업을 손쉽게 반복할 수 있습니다. 2. 환경 설정 2-1. CLOVA Studio API 준비 CLOVA Studio에서는 Chat Completions, 임베딩을 비롯한 주요 API에 대해 OpenAI API와의 호환성을 지원합니다. 본 예제에서는 OpenAI 호환 API 중 Chat Completions 엔드포인트(/chat/completions)를 활용하며, 상세 호환 정보는 OpenAI 호환성 가이드를 참고하시기 바랍니다. 또한, 해당 API 호출하려면 CLOVA Studio에서 발급받은 API 키가 필요합니다. API 키 발급 방법은 CLOVA Studio API 가이드에서 확인할 수 있습니다. 2-2. 프로젝트 구성 프로젝트의 전체 파일 구조는 다음과 같습니다. Python은 3.10 이상을 사용하며, 3.13 버전을 권장합니다. agent_lightning_cookbook/ ├── .env ├── rollout.py ├── run_example.py ├── prompts/ │ ├── apply_edit_ko.poml │ └── text_gradient_ko.poml └── train_apo.py 2-3. 환경 변수 설정 루트 디렉토리에 .env 파일을 생성한 뒤, 앞서 발급받은 API Key를 다음과 같이 입력하고 저장합니다. 이때 따옴표 없이 값을 작성해야 하며, VS Code에서 실행할 경우 설정에서 Use Env File 옵션이 활성화되어 있는지 확인하세요. CLOVA_STUDIO_API_KEY=YOUR_API_KEY 2-4. 패키지 설치 프로젝트에 필요한 패키지를 다음 코드를 실행하여 설치합니다. pip install agentlightning openai python-dotenv poml 3. Rollout 구현 Agent Lightning의 롤아웃 구조를 단일 파일로 단순화해 구현해 봅니다. 본 예제에서는 자연어 요청을 5개 카테고리(주행, 차량 상태, 차량 제어, 미디어, 생활정보)로 분류하는 에이전트를 구성합니다. 각 태스크는 @dataclass로 정의되며, CLOVA Studio의 HCX-005 모델을 사용해 분류를 수행합니다. 시스템 프롬프트에는 다섯 가지 카테고리의 정의와 출력 규칙이 포함되어 있으며, 모델은 입력 문장을 읽고 리스트 형태의 문자열로 응답합니다. 정답은 하나 또는 여러 개일 수 있으며, 어떤 카테고리에도 해당하지 않는 경우에는 빈 리스트([])를 반환하는 것이 올바른 출력입니다. run_rollout() 함수는 한 번의 태스크 실행 단위를 나타내며, 태스크를 실행하고 그 결과를 기반으로 보상을 계산하는 역할을 합니다. 보상은 다음 규칙에 따라 계산되며, 이는 서비스 목적에 따라 자유롭게 커스터마이즈하고 확장할 수 있습니다. 완전 일치(1.0): 모델 응답과 정답이 형식과 내용까지 모두 정확히 일치하는 경우 부분 일치 형식 불일치(0.9): 내용(레이블 집합)이 완전히 동일하지만, 따옴표, 공백, 대소문자 등의 형식이 다른 경우 부분 문자열 일치(0.5): 레이블이 완전히 같지는 않지만 문자열이 부분적으로 겹치는 경우(예: '주행'과 '차량 주행') 부분 레이블 일치(0.5): 여러 레이블 중 일부만 맞힌 경우(예: 2개 중 1개만 맞힌 경우) 불일치(0.0): 리스트 형태로 파싱할 수 없거나, 파싱되더라도 정답과의 교집합이 전혀 없는 경우 이렇게 계산된 보상은 emit_reward()를 통해 Agent Lightning 내부에 기록되며, 이후 APO가 프롬프트를 개선할 때 신호로 활용될 수 있습니다. 아래 코드는 태스크를 한 번 실행하고, 모델 응답을 평가해 보상을 기록하는 롤아웃 루프를 단일 파일로 구현한 예제입니다. 이는 프롬프트를 개선할 때 사용하는 핵심 루프 역할을 합니다. 아래 코드를 rollout.py로 저장하세요. # rollout.py import os import re import json from dataclasses import dataclass from typing import Optional, List import asyncio from dotenv import load_dotenv from openai import AsyncOpenAI, RateLimitError from agentlightning import emit_reward load_dotenv() # --- CLOVA Studio 설정 --- BASE_URL = "https://clovastudio.stream.ntruss.com/v1/openai" API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") # --- 태스크 정의 --- @dataclass class Task: """ 분류 태스크를 표현하는 데이터 구조입니다. - question: 분류 대상 문장 - expected_labels: 정답으로 기대하는 레이블 리스트(예: ["주행", "미디어"]) - task_id: (선택) 태스크 식별자 - system_prompt: (선택) 기본 시스템 프롬프트를 덮어쓰고 싶을 때 사용 """ question: str expected_labels: List[str] task_id: Optional[str] = None system_prompt: Optional[str] = None # --- 유틸리티 --- def normalize_list(values: List[str]) -> List[str]: """리스트 값 정규화""" return [v.strip().lower() for v in values] class RewardCalculator: """보상 계산 유틸리티""" @staticmethod def normalize(s: str) -> str: """문자열 정규화: 따옴표 제거, 공백 제거, 소문자 변환""" return s.strip().strip('\'"').lower() @staticmethod def is_partial_match(expected: str, actual: str) -> bool: """부분 문자열 일치 여부 확인""" e = RewardCalculator.normalize(expected) a = RewardCalculator.normalize(actual) return (e in a) or (a in e) # --- LLM 클라이언트 --- class ClovaClient: def __init__(self, model: str = "HCX-005", temperature: float = 0.0): self.client = AsyncOpenAI( base_url=BASE_URL, api_key=API_KEY, ) self.model = model self.temperature = temperature async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): # asyncio.run()이 루프를 닫기 전에 안전하게 정리 try: self.client.close() except Exception: pass return False async def classify(self, task: Task) -> str: system_prompt = task.system_prompt or """ 당신은 분류기입니다. 입력 문장을 아래 5개 카테고리 중 해당되는 항목으로 분류하세요. 카테고리 정의: - 주행: 주행 및 내비게이션 관련 요청 - 차량 상태: 차량 진단/상태 확인 - 차량 제어: 차량 기능 조작 요청 - 미디어: 음악/라디오, 엔터테인먼트 요청 - 개인 비서: 전화, 메시지, 일정 등 개인 비서 기능 요청 출력 포맷: list 형태로 응답합니다. 해당되는 카테고리가 없다면 빈 배열로 응답하세요. 배열 내 문자열은 작은 따옴표로 감싸세요. """ messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": task.question}, ] # LLM 호출(429 에러 시 재시도) max_retries = 5 wait_time = 2 for attempt in range(max_retries): try: resp = await self.client.chat.completions.create( model=self.model, messages=messages, temperature=self.temperature, ) return resp.choices[0].message.content.strip() except RateLimitError: if attempt < max_retries - 1: # 지수 백오프: 2초, 4초, 8초, 16초, 32초 await asyncio.sleep(wait_time) wait_time *= 2 else: raise except Exception: if attempt < max_retries - 1: await asyncio.sleep(wait_time) wait_time *= 2 else: raise # --- 롤아웃 함수 --- async def run_rollout(task: Task, client: ClovaClient) -> tuple[str, float]: """ 단일 롤아웃 실행: 1) 분류 실행 2) 보상 계산 3) 보상 emit """ # LLM 호출 predicted = await client.classify(task) # JSON 파싱을 위한 전처리 predicted_normalized = predicted.replace("'", '"') try: parsed = json.loads(predicted_normalized) if isinstance(parsed, list): predicted_list = [str(x).strip() for x in parsed] elif isinstance(parsed, str): predicted_list = [parsed] elif isinstance(parsed, dict): # JSON 객체인 경우: categories 또는 labels 키 찾기 if "categories" in parsed: categories = parsed["categories"] if isinstance(categories, list): predicted_list = [str(x).strip() for x in categories] else: predicted_list = [str(categories).strip()] elif "labels" in parsed: labels = parsed["labels"] if isinstance(labels, list): predicted_list = [str(x).strip() for x in labels] else: predicted_list = [str(labels).strip()] else: # 다른 구조의 dict -> 0.0 reward = 0.0 try: emit_reward(reward) except RuntimeError: pass return predicted, reward else: # 리스트도 문자열도 dict도 아님 -> 0.0 reward = 0.0 try: emit_reward(reward) except RuntimeError: pass return predicted, reward except Exception: # JSON 파싱 실패 -> 0.0 reward = 0.0 try: emit_reward(reward) except RuntimeError: pass return predicted, reward # 파싱 성공 시 내용 비교 calculator = RewardCalculator() expected_norm = [calculator.normalize(x) for x in task.expected_labels] actual_norm = [calculator.normalize(x) for x in predicted_list] if sorted(expected_norm) == sorted(actual_norm): expected_json = str(task.expected_labels) if predicted.strip() == expected_json: reward = 1.0 # 완전 일치 elif "'" in predicted: reward = 0.9 # 형식 불일치 elif '"' in predicted: reward = 0.9 # 형식 불일치 else: reward = 0.9 # 형식 불일치 elif set(expected_norm) & set(actual_norm): # 일부 레이블만 일치 reward = 0.5 else: # 부분 문자열 일치 확인 has_partial = False for e in expected_norm: for a in actual_norm: if calculator.is_partial_match(e, a): has_partial = True break if has_partial: break reward = 0.5 if has_partial else 0.0 # Agent Lightning에 보상 emit try: emit_reward(reward) except RuntimeError: pass return predicted, reward 다음은 롤아웃 샘플 실행 코드입니다. 아래 코드를 실행하면 에이전트가 다섯 개의 샘플 태스크를 순차적으로 수행하며, 각 태스크에 대한 모델 응답을 평가하고 보상을 계산·기록하는 과정을 확인할 수 있습니다. # run_example.py import asyncio from rollout import Task, ClovaClient, run_rollout async def run_tests(): # 샘플 태스크 정의 tasks = [ Task( question="회사까지 가장 빠른 길 안내 시작해줘", expected_labels=["주행"], task_id="task_01", ), Task( question="타이어 공기압 체크", expected_labels=["차량 상태"], task_id="task_02", ), Task( question="온열 시트 켜고 출근길에 듣기 좋은 노래 틀어줘", expected_labels=["차량 제어", "미디어"], task_id="task_03", ), Task( question="엄마한테 전화 좀 걸어줘", expected_labels=["개인 비서"], task_id="task_04", ), Task( question="1+1은?", expected_labels=[], task_id="task_05", ), ] client = ClovaClient() for i, task in enumerate(tasks, 1): print(f"[Task {i}/{len(tasks)}] {task.task_id}") print(f"질의: {task.question}") predicted, reward = await run_rollout(task, client) print(f"모델 응답: {predicted}") print(f"실제 정답: {task.expected_labels}") print(f"Reward: {reward:.2f}\n") if __name__ == "__main__": asyncio.run(run_tests()) 위 스크립트 실행 결과입니다. 결과를 보면, 일부 개선이 필요한 태스크를 확인할 수 있습니다. 이러한 부분은 APO를 활용한 프롬프트 자동 최적화를 통해 지침을 점진적으로 정교화함으로써 자연스럽게 개선될 수 있습니다. [Task 1/5] task_01 질의: 회사까지 가장 빠른 길 안내 시작해줘 모델 응답: ['주행'] 실제 정답: ['주행'] Reward: 1.00 [Task 2/5] task_02 질의: 타이어 공기압 체크 모델 응답: ['차량 상태'] 실제 정답: ['차량 상태'] Reward: 1.00 [Task 3/5] task_03 질의: 온열 시트 켜고 출근길에 듣기 좋은 노래 틀어줘 모델 응답: ["차량 제어", "미디어"] 실제 정답: ['차량 제어', '미디어'] Reward: 0.90 [Task 4/5] task_04 질의: 엄마한테 전화 좀 걸어줘 모델 응답: ['개인 비서'] 실제 정답: ['개인 비서'] Reward: 1.00 [Task 5/5] task_05 질의: 1+1은? 모델 응답: [] 실제 정답: [] Reward: 1.00 4. APO 트레이너 APO는 Agent Lightning에 내장된 자동 프롬프트 개선 알고리즘입니다. 에이전트가 여러 태스크를 수행하며 얻은 보상을 기반으로 프롬프트 템플릿을 반복적으로 수정해, 더 높은 성능의 프롬프트로 수렴시키는 방식으로 동작합니다. APO의 최적화 과정은 다음 두 단계로 구성됩니다. Gradient 단계: 무엇이 잘못되었고 어떻게 개선해야 하는지를 분석하는 단계입니다. Apply-Edit 단계: Gradient 단계에서 도출된 개선 방향을 기반으로 실제 프롬프트를 재작성하는 단계입니다. 즉, 모델이 어떤 응답을 생성했고 어떤 보상을 받았는지 분석한 뒤, 그 피드백을 기반으로 더 나은 프롬프트 후보를 생성·실험하는 구조입니다. 4-1. POML 커스터마이징 Agent Lightning에서 사용하는 APO 기본 템플릿은 모두 영문 기반 POML 파일로 제공됩니다. 기본 지시문이 영어 프롬프트 최적화를 전제로 설계되어 있기 때문에, 실제 최적화 과정에서도 모델이 영어 중심의 프롬프트를 생성하는 경향이 있습니다. 따라서 본 문서에서는 Microsoft Agent Lighting 레퍼런스 코드를 참고해, 한국어 프롬프트 최적화에 적합한 커스텀 POML 파일을 직접 구성하고 import하는 방식을 사용합니다. Gradient 단계 이 템플릿은 APO의 첫 번째 단계에서 사용되며, LLM이 프롬프트의 문제점을 찾고, 개선 방향을 생성하는 역할을 수행합니다. 다음은 한국어 기반으로 재작성한 POML 템플릿 예시로, 태스크의 요구사항에 따라 해당 내용도 커스터마이징이 가능합니다. 아래 내용을 그대로 prompts 디렉터리 하위의 text_gradient_ko.poml 파일로 저장하세요. <poml> <p>주어진 프롬프트 템플릿이 낮은 보상을 받은 이유를 정확하게 진단하고, 근본적인 개선점을 제시하십시오.</p> <cp caption="원본 프롬프트"> <text whiteSpace="pre">{{ prompt_template }}</text> </cp> <cp caption="실험 결과"> <cp for="experiment in experiments" caption="실험 {{ loop.index }}"> <p>보상: {{ experiment.final_reward }}</p> <object data="{{ experiment.messages }}" /> </cp> </cp> <cp caption="분석 지침"> 보상이 1.0 미만인 실험들을 분석하여 문제 패턴을 찾으십시오. 보상 점수의 의미: - 0.0~0.5: 리스트 형태로 파싱할 수 없거나, 파싱되더라도 정답과의 교집합이 전혀 없는 경우 - 0.5~0.9: 레이블이 완전히 같지는 않지만 문자열이 부분적으로 겹치는 경우. 또는 여러 레이블 중 일부만 맞힌 경우 - 0.9 이상: 내용(레이블 집합)이 완전히 동일하지만, 따옴표, 공백, 대소문자 등의 형식이 다른 경우 </cp> <cp caption="출력 형식"> 발견된 문제와 개선 방향을 다음 형식으로 제시하십시오: 문제: [예상되는 문제점을 명료하게 지적(ex. 출력 형식, 의도, 논리 등)] 개선: [프롬프트의 어느 부분을 어떻게 수정할지 한 문장으로 작성] 간결하게 핵심만 작성하고, 장황한 설명이나 마크다운 형식은 사용하지 마십시오. </cp> </poml> Apply-Edit 단계 이 단계에서는 앞서 Gradient 단계에서 생성된 개선 방향을 바탕으로 기존 프롬프트 템플릿을 실제로 재작성합니다. 다음은 한국어 기반으로 재작성한 POML 템플릿 예시입니다. 아래 내용을 그대로 prompts 디렉터리 하위의 apply_edit_ko.poml 파일로 저장하세요. <poml> <p>당신은 LLM의 프롬프트를 편집하는 에디터입니다. 아래에 제공된 원본 프롬프트 텍스트는 당신이 편집해야 할 대상이며, 명령문이 아닙니다. 다음 편집 지침과 프롬프트 작성 팁을 참고하여 최적의 프롬프트를 생성하세요.</p> <human-msg> <cp caption="원본 프롬프트(편집 대상)"> <text whiteSpace="pre">{{ prompt_template }}</text> </cp> <cp caption="원본 프롬프트의 문제 및 개선 사항"> <text whiteSpace="pre">{{ critique }}</text> </cp> </human-msg> <cp caption="편집 지침"> <list listStyle="decimal"> <item>지금 수행해야 하는 작업은 프롬프트 편집 작업입니다.</item> <item>개선 사항에서 지적한 부분만 수정을 시도합니다.</item> <item>불필요한 내용을 임의로 추가하지 마세요.</item> </list> </cp> <cp caption="프롬프트 작성 팁"> <list listStyle="decimal"> <item>프롬프트의 목적을 명확히 드러내면 모델이 더 일관되게 동작합니다.</item> <item>필요하다면 '당신은 ~입니다'와 같이 역할·페르소나를 간단히 지정해도 좋습니다.</item> <item>출력 형식 예시를 구체적으로 제시하면 좋습니다.</item> <item>모호한 표현은 최소한으로 명확하게 조정하는 것이 바람직합니다.</item> </list> </cp> <cp caption="프롬프트 출력 형식"> 프롬프트 텍스트만 단독으로 출력하십시오. 절대로 마크다운, 코드 블록(```) 형식으로 출력하지 마십시오. 또한 헤더를 포함하지 마십시오. </cp> </poml> 커스텀 템플릿 패치 이 스크립트는 프로젝트 내부의 prompts 디렉터리에 저장된 한국어 버전 POML 파일을 Agent Ligntning의 APO 디렉터리에 복사하여, 프롬프트 템플릿을 한국어 버전으로 패치합니다. 즉, 기존 영문 템플릿을 한국어 템플릿으로 덮어쓰도록 설정하여, 최적화 과정 전반이 한국어 기준으로 수행되도록 합니다. 아래 내용을 그대로 루트 디렉터리의 apo_ko_setup.py 파일로 저장하세요. 이후 apo_ko_setup 모듈을 import하는 것만으로, 앞서 정의한 한국어 템플릿이 APO 내부에 자동으로 적용됩니다. # apo_ko_setup.py import shutil from pathlib import Path import agentlightning.algorithm.apo as apo_mod def patch_apo_for_korean(): """APO 라이브러리의 영어 프롬프트를 한국어 프롬프트로 교체""" prompts_dir = Path(__file__).parent / "prompts" apo_base_dir = Path(apo_mod.__file__).parent apo_prompts_dir = apo_base_dir / "prompts" files = { "text_gradient_ko.poml": "text_gradient_variant01.poml", "apply_edit_ko.poml": "apply_edit_variant01.poml", } if not apo_prompts_dir.exists(): print(f"APO 프롬프트 디렉터리를 찾을 수 없습니다: {apo_prompts_dir}") return for ko_file, apo_file in files.items(): ko_path = prompts_dir / ko_file apo_path = apo_prompts_dir / apo_file if ko_path.exists(): shutil.copy(ko_path, apo_path) else: print(f"{ko_file} 없음") try: patch_apo_for_korean() except Exception as e: print(f"APO 패치 실패: {e}") 4-2. 데이터셋 준비 분류 태스크의 학습 및 평가에 사용할 데이터를 준비합니다. 4-3. 실행 및 결과 아래 코드는 APO 트레이너를 구성하고, 한국어 POML 템플릿을 사용해 프롬프트 최적화 루프를 실행하는 예제입니다. CLOVA Studio의 HCX-005 모델을 기반으로 APO 알고리즘을 초기화하고, 분류 작업에 맞는 초기 프롬프트 템플릿을 initial_resources에 직접 지정하여 학습을 시작합니다. @agl.rollout 데코레이터로 정의된 에이전트는 각 태스크를 실행하면서 LLM 응답을 생성하고, run_rollout에서 계산된 보상 값을 반환합니다. 이 보상 값은 APO가 다음 프롬프트를 수정하고 개선하는 데 핵심적인 학습 신호로 활용됩니다. 트레이너는 초기 프롬프트 템플릿을 기준으로 학습·검증 데이터셋을 반복 실행하며, 보상을 최대화하는 방향으로 프롬프트를 자동으로 수정하고 버전(v0, v1, v2…) 단위로 관리합니다. 학습이 완료되면, trainer.store에 저장된 프롬프트 버전들을 모두 불러와 테스트셋으로 다시 평가합니다. 이 중 가장 높은 성능을 기록한 프롬프트가 최종 버전으로 선택되며, 모든 버전의 프롬프트 내용과 테스트셋 점수는 prompt_history.txt 파일에 저장됩니다. # train_apo.py import os import random import asyncio import logging from copy import deepcopy from dotenv import load_dotenv import agentlightning as agl from openai import AsyncOpenAI from rollout import Task, run_rollout, ClovaClient from dataset import create_classification_dataset import apo_ko_setup # 한국어 POML 패치용 logging.getLogger("agentlightning").setLevel(logging.CRITICAL) load_dotenv() # --- 설정 --- BASE_URL = "https://clovastudio.stream.ntruss.com/v1/openai" API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") MODEL_NAME = "HCX-005" RANDOM_SEED = 42 BEAM_ROUNDS = 1 BEAM_WIDTH = 1 # --- 전역 변수 --- task_counter = 0 @agl.rollout async def classification_agent(task: dict, prompt_template: agl.PromptTemplate) -> float: """ APO에서 호출되는 분류 에이전트. - task: 데이터셋에서 전달된 태스크(dict) - prompt_template: APO가 현재 시점에 사용 중인 시스템 프롬프트 템플릿 """ global task_counter try: task_obj = Task(**task) # APO가 최적화한 프롬프트를 system_prompt로 주입 if hasattr(prompt_template, "template"): prompt_str = prompt_template.template else: prompt_str = str(prompt_template) task_obj.system_prompt = prompt_str # ClovaClient는 컨텍스트 매니저로 사용하여 연결 정리 async with ClovaClient(model=MODEL_NAME) as client: _, reward = await run_rollout(task_obj, client) task_counter += 1 print(f"\r학습 중... (진행: {task_counter})", end="", flush=True) return reward except Exception as e: print(f"\nTask 오류: {e}") return 0.0 async def evaluate_prompt_on_dataset(prompt_template, dataset_tasks): """ 주어진 프롬프트 템플릿으로 데이터셋 평가. APO가 만든 프롬프트(각 버전)에 대해 test셋에서 평균 reward 계산. """ if hasattr(prompt_template, "template"): prompt_str = prompt_template.template else: prompt_str = str(prompt_template) rewards = [] async with ClovaClient(model=MODEL_NAME) as client: for task_item in dataset_tasks: try: task_obj = deepcopy(task_item) if isinstance(task_item, Task) else Task(**task_item) task_obj.system_prompt = prompt_str _, reward = await run_rollout(task_obj, client) rewards.append(reward) except Exception: rewards.append(0.0) return sum(rewards) / len(rewards) if rewards else 0.0 def extract_version_info(trainer_store): """ Trainer.store 내부에서 버전별 프롬프트를 추출. InMemoryLightningStore의 _resources를 직접 읽어서 v0, v1, v2 ... 버전별 prompt_template를 모은다. """ resources_dict = trainer_store._resources if hasattr(trainer_store, "_resources") else {} initial_prompt = None resources_prompts = {} for version in sorted(resources_dict.keys(), key=lambda v: int(v[1:])): resources_update = resources_dict[version] if not (hasattr(resources_update, "resources") and "prompt_template" in resources_update.resources): continue prompt = resources_update.resources["prompt_template"] resources_prompts[version] = prompt if version == "v0": initial_prompt = prompt return { "resources_dict": resources_dict, "resources_prompts": resources_prompts, "initial_prompt": initial_prompt, } def main(): # --- 데이터셋 분할 --- all_tasks = create_classification_dataset() random.seed(RANDOM_SEED) random.shuffle(all_tasks) total = len(all_tasks) train_tasks = all_tasks[: int(total * 0.6)] val_tasks = all_tasks[int(total * 0.6) : int(total * 0.8)] test_tasks = all_tasks[int(total * 0.8) :] # --- APO 설정 --- try: client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY) algorithm = agl.APO( client, gradient_model=MODEL_NAME, apply_edit_model=MODEL_NAME, beam_rounds=BEAM_ROUNDS, beam_width=BEAM_WIDTH, ) except Exception as e: print(f"오류: {e}") return trainer = agl.Trainer( algorithm=algorithm, strategy=agl.SharedMemoryExecutionStrategy(main_thread="algorithm"), tracer=agl.OtelTracer(), initial_resources={ "prompt_template": agl.PromptTemplate( template=""" 당신은 분류기입니다. 입력 문장을 아래 5개 카테고리 중 해당되는 항목으로 분류하세요. 카테고리 정의: - 주행: 주행 및 내비게이션 관련 요청 - 차량 상태: 차량 진단/상태 확인 - 차량 제어: 차량 기능 조작 요청 - 미디어: 음악/라디오, 엔터테인먼트 요청 - 개인 비서: 전화, 메시지, 일정 등 개인 비서 기능 요청 출력 포맷: list 형태로 응답합니다. 해당되는 카테고리가 없다면 빈 배열로 응답하세요. 배열 내 문자열은 작은 따옴표로 감싸세요. """, engine="f-string", ) }, adapter=agl.TraceToMessages(), ) # --- 학습 실행 --- trainer.fit(agent=classification_agent, train_dataset=train_tasks, val_dataset=val_tasks) # --- 버전별 프롬프트 추출 --- if not (hasattr(trainer.store, "_resources") and trainer.store._resources): print("리소스 없음") return info = extract_version_info(trainer.store) if not info["initial_prompt"] or not info["resources_prompts"]: print("프롬프트 추출 실패") return # --- 테스트셋 평가 --- async def run_evaluation(): version_test_results = {} for version in sorted(info["resources_prompts"].keys(), key=lambda v: int(v[1:])): prompt = info["resources_prompts"][version] score = await evaluate_prompt_on_dataset(prompt, test_tasks) version_test_results[version] = score return version_test_results try: version_test_results = asyncio.run(run_evaluation()) # 최적 버전 선택 best_version = max(version_test_results.keys(), key=lambda v: version_test_results[v]) best_score = version_test_results[best_version] initial_test_score = version_test_results.get("v0", 0.0) print("\n" + "=" * 60) print("최종 평가 결과") print("=" * 60) print(f" 초기 프롬프트(v0): {initial_test_score:.3f}") print(f" 수정된 프롬프트({best_version}): {best_score:.3f}\n") # --- 프롬프트 히스토리 저장 --- with open("prompt_history.txt", "w", encoding="utf-8") as f: f.write("=" * 80 + "\n프롬프트 최적화 이력\n" + "=" * 80 + "\n\n") for version in sorted(info["resources_prompts"].keys(), key=lambda v: int(v[1:])): prompt = info["resources_prompts"][version] prompt_str = prompt.template if hasattr(prompt, "template") else str(prompt) score = version_test_results[version] f.write(f"[{version}] 테스트셋 점수: {score:.3f}\n") f.write("-" * 80 + "\n") f.write(f"{prompt_str}\n") f.write("=" * 80 + "\n\n") print("✓ prompt_history.txt 저장\n") except Exception as e: print(f"평가 중 오류: {e}") import traceback traceback.print_exc() if __name__ == "__main__": main() 다음은 위 코드를 실행했을 때 출력된 결과입니다. 동일한 테스트셋을 기반으로 비교한 결과, 수정된 프롬프트(v4)가 기존 프롬프트 대비 더 높은 분류 성능을 보여줌을 확인할 수 있습니다. ============================================================ 최종 평가 결과 ============================================================ 초기 프롬프트(v0): 0.835 수정된 프롬프트(v4): 0.945 ✓ prompt_history.txt 저장 다음은 APO 알고리즘을 통해 자동으로 개선된 프롬프트 v4의 원본입니다. 이후 필요에 따라 학습 파라미터를 조정해 추가적인 최적화 실험을 진행할 수도 있습니다. 문장 분류기를 위한 분류 작업을 수행해주세요. 분류할 카테고리는 다음과 같습니다: - 주행: 주행 및 내비게이션과 관련된 내용 - 차량 상태: 차량 진단이나 상태에 대한 정보 요청 - 차량 제어: 차랑 기능 조작 요청 - 미디어: 음악 또는 라디오 등의 엔터테인먼트 요청 - 개인 비서: 전화걸기, 메시지 보내기, 일정 등록 등과 같은 개인 비서 업무 요청 제공된 문장을 위의 5가지 카테고리 중 가장 적합하다고 생각하는 곳으로 분류하고 그 결과를 list 형태로 제시해 주세요. 예를 들어 아래와 같이 나타낼 수 있습니다: ```plaintext ['주행', '차량 제어'] ``` 위 예시처럼 해당하는 카테고리를 작은따옴표 안에 넣어서 list로 구성해주시면 됩니다. 만약 문장이 어떠한 카테고리에도 속하지 않는다면 빈 배열 `[]`을 반환하셔도 됩니다. 5. 맺음말 이번 쿡북에서는 CLOVA Studio 모델을 기반으로 롤아웃을 구성하고, APO를 통해 프롬프트를 자동으로 개선하는 전체 흐름을 살펴보았습니다. 단순한 분류 태스크도 보상 구조만 잘 설계하면 원하는 방향으로 모델을 안정적으로 유도할 수 있고, APO는 이 보상 신호를 활용해 프롬프트를 점진적으로 더 좋은 형태로 다듬어 줍니다. 이 구조는 다른 도메인의 에이전트에도 그대로 확장할 수 있는데요. 서비스 요구사항에 맞게 태스크, 보상 체계 등을 커스터마이즈하면, 보다 복잡한 워크플로우나 실제 서비스 환경에서도 안정적인 프롬프트를 자동으로 구축할 수 있습니다. 특히 프롬프트 성능이 곧 모델 품질로 이어지는 LLM 기반 시스템에서는, 이러한 자동 최적화가 품질을 빠르게 끌어올리는 데 효과적입니다. 이제 여러분의 서비스 맥락에 맞는 태스크를 넣어 보며 프롬프트가 어떻게 진화하는지 확인해 보세요. 🧐

안녕하세요. 같은 문제를 겪고 있는데 혹시 해결 하셨을까요?

Authorization failed: [401] Unauthorized client가 납니다. Maps -> Application -> 등록 -> Dynamic Map으로 하고 local.properties에도 넣어놨고 혹시 몰라 <meta-data 에도 넣어놨는데 계속 401이 뜹니다.

Authorization failed: [401] Unauthorized client가 납니다. Maps -> Application -> 등록 -> Dynamic Map으로 하고 local.properties에도 넣어놨고 혹시 몰라 <meta-data 에도 넣어놨는데 계속 401이 뜹니다.

조금 더 해봤는데 1. geoserver 어플리케이션에서 자체적으로 래스터 혹은 gpkg와 좌표계 조정해주는 기능이 있어서 초정밀 수준이 아니라면 이 기능으로 해결이 됩니다. 2. 다만 지오서버 벡터/레스터 운영 자체가 난이도가 있었습니다. 3. 위에 반투명 png 레스터 레이어를 올린 상태에서 마커 등의 조작은 원할하게 작동합니다. 저 주제가 아니라 다른 주제로 한건데 이쪽 링크를 참고하시면 좋을 것 같습니다 😄 URL 삼시세끼 모범밥상

안녕하세요. NAVER Maps JavaScript API v3를 사용 중인데 특정 환경에서만 발생하는 오류가 있어 문의드립니다. Issue와 실제 코드를 기반으로 AI가 작성 후 내용검수를 하였습니다. 어색한 부분이 있으면 양해 부탁드립니다. [증상] 네이버 지도 초기화 시 TypeError: Cannot read properties of undefined (reading 'projection') 오류가 환경에 따라 간헐적으로 발생합니다. - macOS Chrome: 처음 발생 후 현재는 정상 동작 - iOS WebView (iOS 26.1): 간헐적 오류 발생 - Staging 환경: 지속적으로 에러 발생 - Production 환경: 동일 기기, 동일 iOS 버전에서 정상 작동 - 재현성: 동일 코드에서도 발생할 때가 있고 안 할 때도 있음 (예측 불가) [에러 로그] 네이버 지도 초기화 중 오류: TypeError: Cannot read properties of undefined (reading 'projection') at new x.MapOptions (maps.js?ncpKeyId=xxxxx&submodules=panorama:11:13158) at x.Map._initMapOptions (maps.js?ncpKeyId=xxxxx&submodules=panorama:11:22770) at new x.Map (maps.js?ncpKeyId=xxxxx&submodules=panorama:11:15892) at initializeMap (naver-map.vue:1856:11) [환경 정보] - SDK 버전: NAVER Maps JavaScript API v3 - 스크립트 URL: https://oapi.map.naver.com/openapi/v3/maps.js?ncpKeyId=xxxxx&submodules=panorama - 프레임워크: Vue 3 + Quasar (TypeScript) - 문제 발생 환경: iOS 26.1 WebView, macOS Chrome (간헐적) [관련 코드] (스크립트 로드) const createScript = (resolve, reject) => { const script = document.createElement('script'); script.type = 'text/javascript'; script.src = `https://oapi.map.naver.com/openapi/v3/maps.js?ncpKeyId=${NAVER_MAP_CLIENT_ID}&submodules=panorama`; script.async = true; script.onload = () => { resolve(); }; document.head.appendChild(script); }; (지도 초기화) const initializeMap = () => { try { if (!naver || !naver.maps) { throw new Error('네이버 지도 API가 로드되지 않았습니다.'); } if (!naverMapRef.value) { throw new Error('지도 컨테이너 DOM 요소를 찾을 수 없습니다.'); } const mapOptions = { center: base, draggable: draggable, zoom: zoom, minZoom: minZoom, maxZoom: maxZoom, padding: padding, zoomControl: zoomControl, zoomControlOptions: { style: naver.maps.ZoomControlStyle.LARGE, position: naver.maps.Position.RIGHT_CENTER, }, mapTypeId: naver.maps.MapTypeId.NORMAL, mapTypeControl: mapTypeControl, mapTypeControlOptions: { mapTypeIds: ['normal', 'terrain'], style: naver.maps.MapTypeControlStyle.DROPDOWN, position: naver.maps.Position.TOP_RIGHT, }, mapDataControl: false, }; map = new naver.maps.Map(naverMapRef.value, mapOptions); } catch (error) { logger.error('네이버 지도 초기화 중 오류:', error); } }; [재현 단계] 1. 지도 컴포넌트 마운트 2. loadNaverMapScript() 호출 → 스크립트 로드 완료 3. initializeMap() 호출 → new naver.maps.Map() 실행 4. SDK 내부 MapOptions 생성 과정에서 projection 속성 접근 시 undefined 오류 [참고: 유사 사례] [topic/412] map panorama 생성시 오류가 발생합니다 - https://www.ncloud-forums.com/topic/412/ - 동일한 패턴: 간헐적 오류 발생 - "같은 위치라도 에러가 발생할 때도 있고 발생하지 않을 때도 있었습니다" - 우회 해결: "파노라마 맵을 초기에 로드하여 보이지 않는 상태로 놔둔 후..." 방식 [질문] 1. SDK 내부에서 projection이 undefined가 되는 조건이 있을까요? (예: WebGL 미지원, 특정 iOS 버전 등) 2. iOS WebView 환경에서 권장되는 초기화 방식이 별도로 있을까요? 감사합니다.

안녕하세요, @태훈2님, 현재 이미지를 포함한 학습은 지원하지 않고 있습니다. 감사합니다.

HCX-005 튜닝 시, 이미지를 포함한 학습은 불가한가요? 오로지 텍스트만 가능한지요? api가이드 문서에 따로 방법이 나와있지 않아서, 여쭙습니다.

공식 문의 채널로 접수해주시면 담당자가 확인 후 답변 드릴 예정입니다. - https://www.ncloud.com/support/question/service

“Maps 상품 이용 중이고, Maps → Application → DISTANCE-CHECK 앱의 변경 화면에서 Geocoding, Directions 5 선택되어 있습니다. 이 Client ID로 map-geocode/v2/geocode 호출 시 errorCode 210 Permission Denied – A subscription to the API is required 가 계속 발생합니다. 해당 Client ID의 권한 설정/로그 확인 부탁드립니다.”

들어가며 실제 서비스 운영 단계에서 MCP 서버를 안정적으로 관리하고, 다양한 환경에서 효율적으로 확장할 수 있는 방법을 이해하는 것이 중요합니다. 또한 CLOVA Studio 모델을 기반으로 손쉽게 MCP 연동 프로토타입을 구현하는 방법도 함께 익혀두면 좋습니다. 마지막 4부에서는 MCP Inspector를 활용한 서버 디버깅, Flowise를 통한 시각적 운영 및 테스트, 그리고 MCP 서버를 작성하는 가이드를 다룹니다. 이러한 실전 팁을 통해 HyperCLOVA X와 MCP를 함께 활용해 더 유연하고 효율적인 서비스 환경을 구축해 볼 수 있습니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 링크 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 링크 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 1. MCP Inspector로 서버 검증하기 MCP Inspector는 Anthropic에서 제공하는 MCP 서버 테스트·디버깅 도구입니다. GUI 환경에서 서버와 연결하여 도구 목록 확인, 도구 호출 테스트, 스트리밍 이벤트 모니터링 등을 쉽게 수행할 수 있습니다. 아래 명령어를 터미널에서 실행하면 브라우저가 자동으로 열립니다. npx @modelcontextprotocol/inspector 브라우저가 열리면 아래 절차에 따라 간단하게 MCP Inspector를 살펴볼 수 있습니다. 서버 연결 (1부) LangChain에서 네이버 검색 도구 연결하기에서 만든 MCP 서버를 연결합니다. 좌측 패널에서 다음의 값을 입력한 후, Connect 버튼을 클릭합니다. Transport Type: Streamable HTTP URL: http://127.0.0.1:8000/mcp/ 도구 목록 확인 상단 메뉴에서 Tools를 선택하고 [List Tools] 버튼을 클릭합니다. 등록된 도구 목록이 표시되며, MCP 서버에 정의된 도구들을 확인할 수 있습니다. 도구 호출 테스트 도구 목록에서 도구를 선택한 후, 우측 입력 영역에 요청 파라미터를 입력하고 실행합니다. 응답이 우측 패널에 실시간으로 표시됩니다. 다음 이미지는 MCP Inspector에서 서버 연결 후 도구 목록을 조회하고, 특정 도구를 실행한 화면 예시입니다. MCP Inspector는 이러한 단순한 호출 테스트 외에도 리소스 조회, 스트리밍 이벤트 모니터링 등의 기능을 지원하므로, MCP 서버를 다양한 방식으로 검증하고 디버깅하는 데 활용할 수 있습니다. 2. Flowise로 노코드 클라이언트 구성하기 Flowise는 MCP 서버와 연결하여 시각적 인터페이스에서 도구를 구성하고 실행할 수 있는 노코드 기반 클라이언트입니다. 기본적으로 MCP 서버와 HTTP 기반으로 통신하며, STDIO로 연결하려면 로컬 환경에 설치된 Flowise로 실행해야 합니다. 또한 Flowise에서는 OpenAI 호환 모델을 연결할 수 있으며, CLOVA Studio의 HCX-005 모델을 이용할 수 있습니다. 단, HCX-007 모델은 현재 일부 파라미터가 호환되지 않아서 연동이 불가능합니다. Flowise 계정 생성 Flowise 예제를 실행하려면 계정 생성이 필요합니다. 다만, Flowise는 필수 구성 요소가 아니며, 선택적으로 활용할 수 있습니다. 클라우드 사용: Flowise에 가입하여 계정을 생성하면 바로 이용할 수 있습니다. 로컬 설치: Node.js 환경에서 다음 명령어로 설치 후 실행할 수 있습니다. 자세한 내용은 Flowise GitHub 문서를 참고해 주세요. npm install -g flowise npx flowise start 워크플로우 생성 절차 다음 절차를 따라서 워크플로우를 생성합니다. Agentflows 시작 Agentflows에서 우측 상단의 [+ Add New] 버튼을 클릭합니다. 노드 추가 좌측 상단의 버튼을 클릭하여 Agent 노드를 드래그하여 추가합니다. 동일한 방법으로 Direct Reply 노드를 드래그하여 추가합니다. 노드 연결 Start, Agent, Direct Reply 순으로 드래그하여 연결합니다. Agent 노드 설정 Agent 노드를 더블 클릭해 노출되는 모달에서 아래 값을 입력합니다. Model: ChatOpenAI Custom ChatOpenAI Custom Parameters Connect Credential: Create New 선택 후 Credential Name을 입력하고 OpenAI API Key에는 CLOVA Studio에서 발급받은 API Key 입력 Model Name: HCX-005 BasePath: https://clovastudio.stream.ntruss.com/v1/openai Messages Message1 Role: System Content: 당신은 친절한 AI 어시스턴트입니다. 사용자의 질문에 대해 신뢰할 수 있는 정보만 근거로 삼아 답변하세요. Message2 Role: User Content: {{question}} Tools Tool: Custom MCP Custom MCP Parameters MCP Server Config: 로컬 환경에서 Flowise를 실행하는 경우에는 아래 값을 입력합니다. 만약 Flowise 클라우드(웹) 환경에서 실행한다면, ngrok 등을 통해 외부 접근이 가능한 URL을 설정해야 합니다. { "url": "http://127.0.0.1:8000/mcp/", "headers": { "Accept": "application/json, text/event-stream", "Content-Type": "application/json" } } Avaliable Actions: 우측의 새로고침 아이콘 클릭한 뒤 도구 선택 Direct Reply 노드 설정 Direct Reply 노드를 더블 클릭해 노출되는 모달에서 아래 값을 입력합니다. {{ agentAgentflow_0 }} 워크플로우 저장 및 실행 우측 상단의 저장 아이콘을 클릭해 워크플로우를 저장합니다. 메시지 아이콘을 클릭해 대화를 실행합니다. 워크플로우 실행 화면 3. MCP 도구 파라미터 설명과 타입 정의하기 파라미터 타입 정의 PEP 484 및 604 타입 힌트로 각 파라미터의 타입을 명시할 수 있습니다. 파이썬 기본 타입부터 일반적인 Pydantic 호환 타입을 대부분 사용할 수 있습니다. from fastmcp import FastMCP from typing import Literal mcp = FastMCP("Example") @mcp.tool def your_tool( param_1: str, # 문자열 param_2: str | int, # 문자열 또는 정수 param_3: Literal["A", "B"], # "A" 또는 "B"만 허용 param_4: list[str] | None # 문자열 리스트 또는 None ) -> dict: # ... return {"ok": True} 필수 및 선택 파라미터 파라미터에 기본값이 없으면 필수(required), 기본값이 있으면 선택(optional) 인자로 스키마에 정의됩니다. LLM이 이 스키마를 참고해 호출 인자를 구성합니다. from fastmcp import FastMCP from typing import Literal mcp = FastMCP("Example") @mcp.tool def your_tool( # 기본값이 없으므로 스키마에서 required로 포함됨(LLM이 반드시 생성해야만 함) param_1: str, # 필수 파라미터: 문자열. 기본값 없음 # 기본값이 있으므로 required 아님. 모든 문자열 또는 정수 허용되며, 미지정 시 "auto" param_2: str | int = "auto", # 선택 파라미터: string | int. 기본값 "auto" # 기본값이 없으므로 스키마에서 required로 포함됨. 값은 "A" 또는 "B" 중 하나만 허용 param_3: Literal["A", "B"], # 필수 파라미터: "A" 또는 "B"만 허용. 기본값 없음 # 기본값이 있으므로 required 아님. 문자열 리스트 또는 None 허용, 미지정 시 None param_4: list[str] | None = None # 선택 파라미터: list[str] 또는 None ) -> dict: # ... return {"ok": True} 파라미터 역할 정의 파라미터별 역할을 정의하기 위해, typing.Annotated를 사용할 수 있습니다. 이후 도구 스키마의 description 필드로 포함되어, LLM이 각 파라미터의 의미를 더 정확히 이해할 수 있도록 돕습니다. from fastmcp import FastMCP from typing import Literal from typing import Annotated mcp = FastMCP("Example") @mcp.tool def your_tool( param_1: Annotated[str, "파라미터 설명 1"], param_2: Annotated[str | int, "파라미터 설명 2"] = "auto", param_3: Annotated[Literal["A", "B"], "파라미터 설명 3"], param_4: Annotated[list[str] | None, "파라미터 설명 4"] = None ) -> dict: # ... return {"ok": True} 특정 파라미터 제외 exclude_args를 사용하면 특정 파라미터를 도구 스키마에서 제외하여 LLM이 해당 값을 생성하지 않도록 할 수 있습니다. 주로 user_id, auth_token처럼 LLM에 노출되면 안되거나, 호출 때마다 서버가 직접 주입해야 하는 값에 사용합니다. from fastmcp import FastMCP from typing import Literal from typing import Annotated mcp = FastMCP("Example") @mcp.tool( name="your_tool", exclude_args=["param1"] ) def your_tool( param_1: Annotated[str, "파라미터 설명 1"], param_2: Annotated[str | int, "파라미터 설명 2"] = "auto", param_3: Annotated[Literal["A", "B"], "파라미터 설명 3"], param_4: Annotated[list[str] | None, "파라미터 설명 4"] = None ) -> dict: # ... return {"ok": True} 4. 개인화 도구 설계하기 (3부) MCP 실전 쿡북: OAuth 인증이 통합된 MCP 서버 구축하기에서 설명한 대로, 사용자가 OAuth 인증을 완료하면 MCP 서버가 액세스 토큰을 안전하게 저장하고, 이후 AI 어시스턴트를 통한 요청이 들어올 때마다 인증 미들웨어가 자동으로 해당 사용자의 토큰을 찾아 제공합니다. 개발자는 복잡한 인증 과정을 신경 쓸 필요 없이 get_access_token()만 호출하면 되며, 이 토큰은 Bearer {token} 형식으로 API 요청에 자동으로 포함됩니다. 파이썬의 contextvars 덕분에 여러 사용자가 동시에 요청해도 토큰이 섞이지 않아 안전합니다. 이 구조를 활용하면 OAuth 토큰을 안전하게 재사용하면서 사용자별 맞춤 기능을 쉽게 추가할 수 있으며, 네이버뿐만 아니라 구글, 카카오, GitHub 등 OAuth를 지원하는 모든 개인화 서비스에 동일한 방식으로 MCP 서버를 구성할 수 있습니다. import os from fastmcp import FastMCP from naver import NaverProvider from dotenv import load_dotenv load_dotenv() # 환경 변수 로드 NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID") NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET") BASE_URL = os.getenv("BASE_URL") # OAuth Provider 초기화 auth = NaverProvider( client_id=NAVER_CLIENT_ID, client_secret=NAVER_CLIENT_SECRET, base_url=BASE_URL ) # MCP 서버 생성 mcp = FastMCP(name="Naver Calendar MCP", auth=auth) # 도구 정의 @mcp.tool(description="네이버 캘린더에 일정을 추가합니다.") async def create_calendar_schedule( title: str, start_datetime: str, end_datetime: str ) -> dict: from fastmcp.server.dependencies import get_access_token import urllib.request # 1. 액세스 토큰 가져오기 token = get_access_token() access_token = token.token # 2. API 요청 준비 header = "Bearer " + access_token url = "https://openapi.naver.com/calendar/createSchedule.json" # 3. iCalendar 문자열 생성 schedule_ical = f"""BEGIN:VCALENDAR VERSION:2.0 PRODID:Naver Calendar BEGIN:VEVENT DTSTART;TZID=Asia/Seoul:{start_datetime} DTEND;TZID=Asia/Seoul:{end_datetime} SUMMARY:{title} END:VEVENT END:VCALENDAR""" # 4. API 요청 전송 data = f"calendarId=defaultCalendarId&scheduleIcalString={schedule_ical}" request = urllib.request.Request(url, data=data.encode("utf-8")) request.add_header("Authorization", header) try: response = urllib.request.urlopen(request) return { "success": True, "message": "일정이 추가되었습니다." } except Exception as e: return { "success": False, "error": str(e) } # 서버 실행 if __name__ == "__main__": mcp.run(transport="streamable-http", path="/mcp") 마무리 1부(서버·클라이언트 연결), 2부(세션·컨텍스트 관리), 3부(OAuth 2.0 인증 예시)를 바탕으로 4부에서는 MCP 운영 효율을 높이는 여러 방법을 다뤘습니다. MCP의 특성을 명확히 이해하고, 서비스 목적에 맞게 설계·적용한다면, 보다 안정적이고 확장 가능한 MCP 생태계를 구축할 수 있습니다. 이제 CLOVA Studio 모델을 활용해, 직접 자신만의 MCP 서버 연동 워크플로우를 만들어 보시길 바랍니다. 🚀 MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 링크 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 링크 ✔︎ (4부) 운영과 확장을 위한 다양한 팁

들어가며 (1부) MCP 실전 쿡북: LangChain에서 네이버 검색 도구 연결하기와 (2부) MCP 실전 쿡북: 세션 관리와 컨텍스트 유지 전략에서는 MCP 서버의 기본 구현과 클라이언트로 MCP 서버를 연결하여 활용하는 방법을 다루었습니다. 이번 3부에서는 FastMCP에서 제공하는 인증 시스템을 활용해 MCP 서버를 구축하는 방법을 다룹니다. OAuth 인증이 통합된 MCP 서버 방식에서는 사용자가 처음 한 번만 네이버 로그인을 진행하면, 서버가 자동으로 토큰을 받아오게 됩니다. 사용자는 토큰이 어떻게 발급되고 유지되는지 신경 쓸 필요 없이, 단순 로그인만으로 자신의 네이버 캘린더나 카페 등 로그인 기반 서비스를 바로 사용할 수 있습니다. 토큰이 만료되더라도 MCP 서버가 자동으로 검증 및 재발급 절차를 처리하므로, 로그인 과정이 반복되지 않습니다. 이를 통해 사용자 계정 기반 외부 서비스와 HyperCLOVA X 모델 간의 안전한 연동이 가능해지고, 개인화된 기능을 더 쉽게 활용할 수 있습니다. 본 쿡북에서는 네이버 로그인 인증을 직접 구현해보고, 추가로 네이버 캘린더까지 연동한 실습 시나리오를 통해 OAuth 인증 기반 MCP 서버의 구축 방법을 단계별로 안내합니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 링크 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 링크 1. 사전 준비 사항 본 섹션에서는 인증 기반 MCP 서버를 구현하기 전에 필요한 사전 준비 사항에 대해 안내합니다. Ngrok 설치 및 등록 Ngrok은 로컬 서버를 외부에서 접근할 수 있도록 해주는 터널링 도구입니다. 본 예제에서는 서버가 로컬 환경에서 실행되기 때문에 외부 접근을 가능하게 하기 위해 Ngrok을 사용했습니다. 실제 서비스 환경에서는 Ngrock 등록 없이 운영 중인 도메인 주소를 그대로 사용해도 됩니다. 먼저 공식 사이트에서 회원가입 후, 안내에 따라 ngrok을 설치하고 발급받은 인증 토큰을 등록해야 합니다. 설치와 설정이 완료되면 터미널에서 다음 명령어를 실행하여, 로컬 8000 포트를 https://xxx.ngrok-free.app/ 형태의 공개 URL로 매핑할 수 있습니다. ngrok http 8000 발급 받은 공개 URL은 이후 환경 변수 설정에 함께 입력합니다. 네이버 개발자 센터 설정 네이버 개발자 센터 > Application > 내 애플리케이션 > Application 등록에서 다음 항목을 설정합니다. 애플리케이션 이름을 설정합니다. 사용 API: '검색','네이버 로그인','캘린더'를 추가합니다. 로그인 오픈 API 서비스 환경 'PC 웹' 환경을 추가합니다. 서비스 URL: MCP 서버의 엔드포인트 주소를 입력합니다.(ex. https://xxx.ngrok-free.app 또는 https://your-domain.com) 네이버 로그인 Callback URL: 서버 주소 뒤에 '/auth/callback'을 추가합니다.(ex. https://xxx.ngrok-free.app/auth/callback 또는 https://your-domain.com/auth/callback) 비로그인 오픈 API 서비스 환경 서비스 URL: MCP 서버의 엔드포인트 주소를 입력합니다.(ex. https://xxx.ngrok-free.app 또는 https://your-domain.com) 페이지 하단의 수정 버튼을 클릭해 설정을 완료합니다. API 호출에 필요한 클라이언트 아이디와 클라이언트 시크릿 정보를 확인합니다. 클라이언트 아이디와 클라이언트 시크릿 확인을 참고해 주세요. 네이버 개발자 센터 > 내 애플리케이션 > 멤버 관리에서 테스트 ID를 등록합니다. 애플리케이션이 '개발 중' 상태이면 멤버 관리 메뉴에서 등록한 아이디만 네이버 로그인을 이용할 수 있습니다. 관리자 아이디 외 다른 아이디로 테스트를 진행할 경우 멤버 관리 메뉴에서 테스트 ID 등록을 진행합니다. 애플리케이션이 '승인'된 상태이면 모든 아이디로 네이버 로그인을 이용할 수 있습니다. 프로젝트 구성 프로젝트의 전체 파일 구조는 다음과 같습니다. Python 버전은 3.10이상, 3.13 미만입니다. mcp_cookbook_part3/ ├── remote_auth_server.py # FastMCP 서버(네이버 OAuth) ├── fastmcp_client.py ├── naver.py # 네이버 OAuth Provider 구현 └── .env 환경 변수 설정 환경 변수를 설정합니다. 루트 디렉터리에 .env 파일을 생성한 뒤, 앞서 발급받은 API Key를 다음과 같이 입력하고 저장합니다. 이때 따옴표 없이 값을 작성해야 하며, VS Code에서 실행할 경우 설정에서 Use Env File 옵션이 활성화되어 있는지 확인하세요. #.env CLOVA_STUDIO_API_KEY=YOUR_API_KEY NAVER_CLIENT_ID=YOUR_CLIENT_ID NAVER_CLIENT_SECRET=YOUR_CLIENT_SECRET BASE_URL=https://xxx.ngrok-free.app # 또는 https://your-domain.com 패키지 설치 프로젝트에 필요한 패키지 목록은 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 루트 디렉터리에 requirements.txt 파일로 저장하세요. requirements.txt 다운로드 루트 디렉터리에서 터미널을 실행하여 다음과 같이 예제 실행에 필요한 패키지를 설치합니다. 가상환경 설치를 권장합니다. # 1. 파이썬 가상환경 생성 python -m venv .venv # 2. 가상환경 활성화 (macOS/Linux) source .venv/bin/activate # (Windows) # .venv/Scripts/activate.ps1 # 3. 패키지 설치 pip install -r requirements.txt 실행 방법 본 쿡북에서는 실행을 위해 총 3개의 터미널이 필요합니다. (Ngrok, 서버, 클라이언트) Ngrok 실행 # 1. 사전 준비 사항에서 설정한 포트로 실행 ngrok http 8000 인증 기반 MCP 서버 실행 # 2. 가상환경 활성화 source .venv/bin/activate # 3. 인증 서버 실행 python remote_auth_server.py FastMCP 클라이언트 실행 # 4. 가상환경 활성화 source .venv/bin/activate # 5. 클라이언트 실행 python fastmcp_client.py 2. MCP 인증 요구 사항 MCP 인증을 구현하기에 앞서, 인증의 기본 개념과 MCP 서버가 준수해야 할 인증 표준 가이드, 그리고 IdP 활용 방안을 소개합니다. 인증(Authentication)과 인가(Authorization) MCP 서버와 같은 외부 클라이언트 요청을 받는 시스템에서는 인증(Authentication)와 인가(Authorization) 체계를 적용해 사용자별 접근 권한을 안전하게 관리하는 것이 중요합니다. 인증(Authentication): 사용자의 신원(ex. 아이디·비밀번호, 소셜 로그인, SSO 등)을 검증하여, 요청자가 실제로 등록된 사용자임을 확인하는 과정입니다. 인가(Authorization): 인증이 완료된 사용자가 접근하려는 리소스나 기능에 대해 권한이 있는지를 판별하는 과정입니다. 이때, 사용자 신원의 확인은 OIDC와 같은 인증 계층이 담당하며, 권한 관리 및 위임 절차는 OAuth 2.0 같은 인가 프레임워크를 통해 진행됩니다. 이 두 기술은 MCP 서버의 핵심 보안 구조를 구성합니다. OIDC를 활용한 인증 계층 OIDC(OpenID Connect)는 OAuth 2.0 프로토콜 위에 구축된 인증 계층입니다. OIDC 인증의 핵심은 ID 토큰(ID Token)을 발급하는 것입니다. 이 토큰은 JWT 형식으로, 사용자의 식별자, 발급 시간, 만료 시간 등의 신원 정보를 담고 있습니다. 클라이언트는 이 ID 토큰을 통해 사용자의 신원을 확인할 수 있게 됩니다. 또한 OIDC는 UserInfo 엔드포인트를 제공해서 사용자 프로필 정보(ex. 이름, 이메일 등)를 조회할 수 있게 합니다. OAuth 2.0 인가 프레임워크 인가를 구현하는 데 가장 널리 사용되는 표준 프레임워크는 OAuth 2.0입니다. 핵심 개념은 클라이언트가 사용자를 대신해 리소스에 접근할 수 있도록 권한을 안전하게 위임하는 구조를 제공한다는 점입니다. 이 프레임워크는 다음 두 구성 요소를 중심으로 동작합니다. 인증 서버(Authorization Server): OIDC 환경에서 사용자의 신원을 확인하고, 클라이언트에게 허용된 스코프(Scope)에 따라 액세스 토큰(Access Token)을 발급할지 결정합니다. 리소스 서버(Resource Server): 실제 리소스(ex. MCP 서버)를 보유하고 있으며, 클라이언트가 전달한 액세스 토큰을 검증해 요청된 작업을 허용할지 결정합니다. MCP 인증 표준 가이드 MCP 공식 가이드(2025-06-18)에서는 견고한 인증 및 인가 환경 구축을 위해 다음 4가지 OAuth 표준을 준수하도록 명시하고 있습니다. OAuth 2.1 IETF DRAFT(draft-ietf-oauth-v2-1-13) OAuth 2.0의 보안 취약점으로 식별된 부분을 해결하고, 모범 사례로 인정된 확장 기능을 필수 사항으로 통합하여 정리한 개정판입니다. 일례로, 모든 클라이언트에 대해 PKCE(Proof Key for Code Exchange)를 필수로 적용하도록 합니다. 자세한 변경 사항은 OAuth 2.1 공식 문서에서 확인할 수 있습니다. 참고로, 현재 OAuth 2.1은 정식 RFC가 아닌 Internet Draft 단계로, 아직 표준화 과정이 진행 중입니다. 그럼에도 불구하고 업계에서는 사실상 가장 안전한 구현 지침으로 받아들여지고 있으며, MCP 서버와 같은 외부 접근 환경에서는 OAuth 2.1을 적용하는 것이 권장됩니다. OAuth 2.0 Authorization Server Metadata(RFC8414) 클라이언트가 OAuth 인증 서버와 상호작용하는 데 필요한 정보를 자동으로 얻을 수 있게 해주는 메타데이터 형식에 대한 표준입니다. 이 메타데이터에는 인증 엔드포인트, 토큰 엔드포인트, 지원하는 스코프, 지원하는 그랜트 타입(Grant Type) 등의 정보가 포함됩니다. 이 표준을 사용하면 MCP 인증 서버의 위치와 기능을 클라이언트가 동적으로 검색하고 구성할 수 있습니다. OAuth 2.0 Dynamic Client Registration Protocol(RFC7591) Dynamic Client Registration(DCR)은 클라이언트가 인증 서버에 자동으로 등록될 수 있게 해주는 표준 방식입니다. 기존에는 클라이언트(애플리케이션)을 사용하기 전, 관리자가 인증 서버에 필요한 정보를 수동으로 등록하고 client_id를 받아야 했지만, 이 표준을 사용하면 이 절차가 간소화됩니다. 클라이언트는 등록 엔드포인트에 자신의 클라이언트 메타데이터를 전송하기만 하면, 서버가 자동으로 client_id와 client_secret을 발급합니다. 즉, MCP 기반 애플리케이션의 자동화된 배포 및 확장을 지원하여, 새로운 클라이언트가 MCP 환경에 빠르게 통합될 수 있도록 합니다. OAuth 2.0 Protected Resource Metadata(RFC9728) 리소스 서버가 자신이 어떤 인증 및 인가 요구 사항을 가지고 있는지 메타데이터로 공개할 수 있도록 하는 표준입니다. 메타데이터에는 토큰 형식, 허용되는 스코프, 특별한 보안 요구 사항이 있는지 등의 정보가 포함됩니다. 이 표준을 통해 클라이언트나 인증 서버는 리소스 서버의 요구 사항을 동적으로 인식할 수 있습니다. IdP와 OAuth Proxy의 활용 MCP 서버에 OAuth 기반 인증 체계를 구축할 때, 기존의 IdP(Identity Provider)를 활용하는 것이 일반적입니다. IdP는 사용자의 신원을 확인하고 인증 서비스를 제공하는 주체입니다. 대표적으로 구글, 네이버와 같은 소셜 로그인 서비스, 그리고 Stytch, WorkOS, Auth0과 같은 기업용 SSO(Single Sign-On) 솔루션이 있습니다. 현재 OAuth 2.1은 정식 RFC 표준으로 확정되지 않은 단계이기 때문에, 대부분의 IdP는 공식적으로 OAuth 2.0을 지원합니다. 따라서 실제 구현 시에는 MCP 표준에서 요구하는 PKCE와 같은 핵심 보안 메커니즘을 지원하는 IdP로 채택하는 것이 좋습니다. 만약 선택한 IdP가 최신 보안 기능을 제공하지 않거나, DCR 또는 메타데이터 기반 동적 구성 기능을 지원하지 않는다면, FastMCP의 OAuth Proxy 같은 어댑터를 연결해 사용할 수 있습니다. FastMCP의 OAuth Proxy는 MCP 규격과 호환되지 않는 외부 로그인 서비스를 MCP 클라이언트가 기대하는 DCR 기반 인터페이스로 변환해 줍니다. 즉, 클라이언트는 MCP 표준 요청을 보내면, OAuth Proxy가 이를 IdP가 처리할 수 있는 형태로 중개하고, 응답은 다시 MCP 형식으로 재구성합니다. 3. OAuth Proxy 기반 인증 커스터마이징 이번 섹션에서는 FastMCP의 기본 코드를 기반으로 네이버 인증을 구현하는 방법을 설명합니다. OAuth Proxy 동작 흐름 FastMCP에서 제공하는 OAuth Proxy는 MCP 클라이언트와 외부 인증 서버 사이에서 어댑터처럼 동작합니다. MCP 클라이언트에는 MCP 표준 인증 서버로 보이도록 /register와 /.well-known/... 엔드포인트를 제공해주고, 네이버 인증 서버에는 미리 등록된 일반 OAuth 클라이언트로 동작하여 요청을 중계합니다. 즉, OAuth Proxy는 인증 서버가 MCP 표준을 지원하지 않아도, 클라이언트 입장에서는 MCP 표준 인증 흐름이 그대로 유지되는 것처럼 보이게 하는 역할을 합니다. PKCE(Proof Key for Code Exchange)란? PKCE는 OAuth 인증 과정에서 인가 코드가 탈취되는 것을 방지하는 보안 메커니즘입니다. 인증을 시작할 때 무작위 보안 코드(code_challenge)를 생성하여 전달하고, 토큰을 요청할 때 원본 검증 코드(code_verifier)를 함께 제출합니다. 인증 서버는 검증 코드가 처음 받은 보안 코드와 일치하는지 확인하여, 동일한 클라이언트가 요청했음을 증명합니다. 이 방식은 중간에 인가 코드를 가로채더라도 검증 코드 없이는 토큰을 발급받을 수 없어 안전합니다. OAuth Proxy의 역할 중 MCP 표준 인증 흐름을 위해 이중 PKCE 검증을 수행합니다. 첫 번째는 MCP 클라이언트와 Proxy 간의 PKCE 검증이고, 두 번째는 Proxy와 실제 OAuth Provider 간의 PKCE 검증입니다. 이를 통해 양쪽 구간 모두에서 보안이 강화됩니다. NaverProvider 구현 다음은 FastMCP에서 공식 제공하는 GoogleProvider를 커스터마이징 하여 NaverProvider를 구현한 스크립트입니다. 레퍼런스 코드는 여기에서 확인할 수 있습니다. # naver.py from __future__ import annotations import httpx from key_value.aio.protocols import AsyncKeyValue from pydantic import AnyHttpUrl, SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict from fastmcp.server.auth import TokenVerifier from fastmcp.server.auth.auth import AccessToken from fastmcp.server.auth.oauth_proxy import OAuthProxy from fastmcp.settings import ENV_FILE from fastmcp.utilities.auth import parse_scopes from fastmcp.utilities.logging import get_logger from fastmcp.utilities.types import NotSet, NotSetT logger = get_logger(__name__) class NaverProviderSettings(BaseSettings): """Settings for Naver OAuth provider.""" model_config = SettingsConfigDict( env_prefix="NAVER_", env_file=ENV_FILE, extra="ignore", ) client_id: str | None = None client_secret: SecretStr | None = None base_url: AnyHttpUrl | str | None = None issuer_url: AnyHttpUrl | str | None = None redirect_path: str | None = None required_scopes: list[str] | None = None timeout_seconds: int | None = None allowed_client_redirect_uris: list[str] | None = None jwt_signing_key: str | None = None @field_validator("required_scopes", mode="before") @classmethod def _parse_scopes(cls, v): return parse_scopes(v) class NaverTokenVerifier(TokenVerifier): """Token verifier for Naver OAuth tokens. 네이버의 사용자 프로필 API를 호출하여 검증해야 합니다. 이 검증기는 네이버 API 사양에 따라 토큰 검증과 사용자 정보 추출을 처리합니다. 주요 특징: - 네이버 프로필 API (/v1/nid/me)를 사용한 토큰 검증 - resultcode/message 구조를 가진 네이버 특유의 응답 형식 처리 - 네이버에서 제공하는 포괄적인 사용자 프로필 데이터 추출 """ def __init__( self, *, required_scopes: list[str] | None = None, timeout_seconds: int = 10, ): """네이버 토큰 검증기를 초기화합니다. Args: required_scopes: 필수 OAuth 스코프. timeout_seconds: API 호출을 위한 HTTP 요청 타임아웃 (기본값: 10초) """ super().__init__(required_scopes=required_scopes) self.timeout_seconds = timeout_seconds async def verify_token(self, token: str) -> AccessToken | None: """네이버 OAuth 토큰을 검증하고 사용자 정보를 추출합니다. 이 메서드는 네이버 프로필 API를 호출하여 토큰을 검증하고, 성공할 경우 사용자 클레임이 포함된 AccessToken을 반환합니다. Args: token: 검증할 액세스 토큰 Returns: 유효한 경우 사용자 클레임이 포함된 AccessToken, 무효한 경우 None """ try: async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: # 네이버 프로필 API 호출 response = await client.get( "https://openapi.naver.com/v1/nid/me", headers={ "Authorization": f"Bearer {token}", "User-Agent": "FastMCP-Naver-OAuth", }, ) # HTTP 상태 코드 확인 if response.status_code != 200: logger.debug( "Naver profile API returned HTTP %d for token verification", response.status_code, ) return None # JSON 응답 파싱 try: api_response = response.json() except Exception as e: logger.debug("Failed to parse Naver API response as JSON: %s", e) return None # 네이버 프로필 API 결과 코드 확인 result_code = api_response.get("resultcode") if result_code != "00": logger.debug( "Naver API returned error code %s: %s", result_code, api_response.get("message", "Unknown error"), ) return None # 'response' 필드에서 사용자 프로필 데이터 추출 user_profile = api_response.get("response") if not user_profile or not isinstance(user_profile, dict): logger.debug("Naver API response missing user profile data") return None # 필수 사용자 ID 검증 user_id = user_profile.get("id") if not user_id: logger.debug("Naver user profile missing required 'id' field") return None # 스코프 검증 # 네이버의 프로필 API는 스코프 정보를 반환하지 않음 # 프로필 API 접근 성공으로 스코프 유효성을 간접 검증 print("required_scopes:",self.required_scopes) validated_scopes = self.required_scopes or ["openid"] if self.required_scopes: if "openid" in self.required_scopes: logger.debug("Naver openid scope validated via successful profile access") else: logger.warning( "Naver scope validation limited - profile API doesn't return scope info" ) # 네이버 프로필 데이터로부터 사용자 클레임 구성 user_claims = { "sub": user_id, "iss": "naver", } # 네이버에서 제공하는 사용자 프로필 필드들 profile_fields = { "email": user_profile.get("email"), "name": user_profile.get("name"), "nickname": user_profile.get("nickname"), "profile_image": user_profile.get("profile_image"), "age": user_profile.get("age"), "gender": user_profile.get("gender"), "birthday": user_profile.get("birthday"), "birthyear": user_profile.get("birthyear"), "mobile": user_profile.get("mobile"), } # 값이 있는 필드만 클레임에 추가 for field, value in profile_fields.items(): if value is not None and value != "": user_claims[field] = value # 원본 프로필 데이터 보존 user_claims["naver_profile"] = user_profile # AccessToken 생성 # Note: 네이버는 프로필 API에서 토큰 만료 정보를 제공하지 않음 # 토큰 교환 시점에서 expires_in 정보를 별도로 추적해야 함 access_token = AccessToken( token=token, client_id="naver-verified", # 네이버 검증된 토큰 식별자 scopes=validated_scopes, expires_at=None, # 토큰 교환 시 설정됨 claims=user_claims, ) logger.debug( "Naver token verified successfully for user %s (nickname: %s)", user_id, user_profile.get("nickname", "unknown") ) return access_token except httpx.RequestError as e: logger.debug("Network error during Naver token verification: %s", e) return None except Exception as e: logger.error( "Unexpected error during Naver token verification: %s", e, exc_info=True ) return None class NaverProvider(OAuthProxy): """FastMCP용 네이버 OAuth 프로바이더. 이 프로바이더는 모든 FastMCP 서버에 네이버 OAuth 보호 기능을 쉽게 추가할 수 있게 합니다. 네이버 OAuth 앱 자격 증명과 기본 URL만 제공하면 바로 사용할 수 있습니다. 주요 기능: - 네이버에 대한 OAuth Proxy - 네이버 사용자 정보 API를 통한 자동 토큰 검증 - 네이버 프로필 API로부터 사용자 정보 추출 - PKCE, DCR 인터페이스, 메타데이터 제공 등 네이버 로그인이 MCP 표준을 간접적으로 준수하도록 기능 제공 사용 예시: ```python from fastmcp import FastMCP from fastmcp.server.auth.providers.naver import NaverProvider auth = NaverProvider( client_id="your_naver_client_id", client_secret="your_naver_client_secret", base_url="https://my-server.com" ) mcp = FastMCP("My App", auth=auth) ``` """ def __init__( self, *, client_id: str | NotSetT = NotSet, client_secret: str | NotSetT = NotSet, base_url: AnyHttpUrl | str | NotSetT = NotSet, issuer_url: AnyHttpUrl | str | NotSetT = NotSet, redirect_path: str | NotSetT = NotSet, required_scopes: list[str] | NotSetT = NotSet, timeout_seconds: int | NotSetT = NotSet, allowed_client_redirect_uris: list[str] | NotSetT = NotSet, client_storage: AsyncKeyValue | None = None, jwt_signing_key: str | bytes | NotSetT = NotSet, require_authorization_consent: bool = True, ): """네이버 OAuth 프로바이더를 초기화합니다. Args: client_id: 네이버 개발자센터에서 발급받은 OAuth 클라이언트 ID. client_secret: 네이버 개발자센터에서 발급받은 OAuth 클라이언트 시크릿. base_url: OAuth 엔드포인트가 접근 가능한 공개 URL (마운트 경로 포함). issuer_url: OAuth 메타데이터를 위한 Issuer URL (기본값: base_url). 경로 하위에 마운트할 때 discovery 중 404 오류를 방지하기 위해 루트 레벨 URL을 사용하세요. redirect_path: 네이버 OAuth 앱에 설정된 리다이렉트 경로 (기본값: "/auth/callback"). required_scopes: 필수 네이버 스코프 (기본값: ["openid"]). 일반적인 스코프: - "openid" - OpenID Connect용 (기본값) - "profile" - 프로필 정보 접근 - "email" - 이메일 접근 timeout_seconds: 네이버 API 호출을 위한 HTTP 요청 타임아웃 (기본값: 10초). allowed_client_redirect_uris: MCP 클라이언트용 허용된 리다이렉트 URI 패턴 목록. None(기본값)인 경우 모든 URI 허용, 빈 목록인 경우 URI 허용 안함. client_storage: OAuth 상태를 위한 스토리지 백엔드 (클라이언트 등록, 암호화된 토큰). None인 경우 데이터 디렉토리(`platformdirs`에서 파생)에 DiskStore가 생성됩니다. 디스크 스토어는 JWT 서명 키에서 파생된 키를 사용하여 암호화됩니다. jwt_signing_key: FastMCP JWT 토큰 서명을 위한 시크릿 (문자열 또는 바이트). 바이트가 제공되면 그대로 사용됩니다. 문자열이 제공되면 32바이트 키로 파생됩니다. 제공되지 않으면 업스트림 클라이언트 시크릿이 PBKDF2를 사용하여 32바이트 키로 파생됩니다. require_authorization_consent: 클라이언트 인증 전 사용자 동의 필요 여부 (기본값: True). True인 경우 네이버로 리다이렉트되기 전에 사용자에게 동의 화면이 표시됩니다. False인 경우 사용자 확인 없이 직접 인증이 진행됩니다. 보안 경고: 로컬 개발 또는 테스트 환경에서만 비활성화하세요. """ settings = NaverProviderSettings.model_validate( { k: v for k, v in { "client_id": client_id, "client_secret": client_secret, "base_url": base_url, "issuer_url": issuer_url, "redirect_path": redirect_path, "required_scopes": required_scopes, "timeout_seconds": timeout_seconds, "allowed_client_redirect_uris": allowed_client_redirect_uris, "jwt_signing_key": jwt_signing_key, }.items() if v is not NotSet } ) # 필수 설정 검증 if not settings.client_id: raise ValueError( "client_id is required - set via parameter or NAVER_CLIENT_ID" ) if not settings.client_secret: raise ValueError( "client_secret is required - set via parameter or NAVER_CLIENT_SECRET" ) # 기본값 적용 timeout_seconds_final = settings.timeout_seconds or 10 required_scopes_final = settings.required_scopes or ['openid'] allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris # 네이버 토큰 검증기 생성 token_verifier = NaverTokenVerifier( required_scopes=required_scopes_final, timeout_seconds=timeout_seconds_final, ) # SecretStr에서 문자열 추출 client_secret_str = ( settings.client_secret.get_secret_value() if settings.client_secret else "" ) # 네이버 엔드포인트로 OAuth Proxy 초기화 super().__init__( upstream_authorization_endpoint="https://nid.naver.com/oauth2/authorize", upstream_token_endpoint="https://nid.naver.com/oauth2/token", upstream_client_id=settings.client_id, upstream_client_secret=client_secret_str, token_verifier=token_verifier, base_url=settings.base_url, redirect_path=settings.redirect_path, issuer_url=settings.issuer_url or settings.base_url, allowed_client_redirect_uris=allowed_client_redirect_uris_final, client_storage=client_storage, jwt_signing_key=settings.jwt_signing_key, require_authorization_consent=require_authorization_consent, token_endpoint_auth_method="client_secret_post", # 본문 요청에 포함하여 전달 ) logger.info( "Initialized Naver OAuth provider for client %s with scopes: %s", settings.client_id, required_scopes_final, ) 위 스크립트를 만들기 위해 레퍼런스 코드에서 토큰 검증 방식과 API 엔드 포인트 구성, 두 가지 항목에 커스터마이징을 적용하였습니다. 각 항목의 구체적인 차이점은 이어서 설명하도록 하겠습니다. 이 과정을 참고하면, 서비스 환경에 맞게 기업별 SSO나 소셜 로그인 등 다양한 인증 방식을 확장 적용할 수 있습니다. 토큰 검증 방식의 차이 각 서비스의 API 명세가 다르기 때문에 토큰의 유효성을 확인하는 방법에 차이가 있습니다. GoogleTokenVerifier 다음은 Google의 tokeninfo API(/oauth2/v1/tokeninfo)를 호출한 예시입니다. 해당 API는 토큰 자체의 정보(만료 시간, 스코프 등)를 직접 반환해 줍니다. 필요에 따라 userinfo API를 추가로 호출하여 사용자 프로필을 가져올 수 있습니다. 구글 tokeninfo API 응답 예시 구글 userinfo API 응답 예시 NaverTokenVerifier 다음은 네이버의 사용자 프로필 API(/v1/nid/me)를 호출한 예시입니다. 해당 API는 토큰 자체의 정보보다는 사용자 프로필을 바로 반환합니다. 따라서, Google과 달리 네이버는 토큰 검증 과정에서 토큰 자체의 정보(만료 시간, 스코프 등)을 알 수 없습니다. 네이버 사용자 프로필 API 응답 예시 API 엔드포인트 및 인증 방식의 차이 OAuthProxy가 어떤 서버와 통신할지를 정의하는 API 엔드포인트와 인증 설정값이 서로 다릅니다. GoogleProvider API 엔드포인트: accounts.google.com, oauth2.googleapis.com 등 Google의 공식 주소를 사용합니다. 인증 방식: token_endpoint_auth_method를 따로 지정하지 않아, 기본 값인 client_secret_basic을 사용합니다.(ID/Secret을 HTTP 헤더에 담아 전송) super().__init__( upstream_authorization_endpoint="https://accounts.google.com/o/oauth2/v2/auth", upstream_token_endpoint="https://oauth2.googleapis.com/token", ... ) NaverProvider API 엔드포인트: nid.naver.com으로 시작하는 네이버의 공식 주소를 사용합니다. 인증 방식: 네이버 API 명세에 따라 token-endpoint-auth_method="client_secret_post"를 명시적으로 설정했습니다.(ID/Secret을 HTTP 요청 본문에 담아 전송) super().__init__( upstream_authorization_endpoint="https://nid.naver.com/oauth2/authorize", upstream_token_endpoint="https://nid.naver.com/oauth2/token", ... token_endpoint_auth_method="client_secret_post" ) 4. OAuth 인증이 통합된 MCP 서버 구현 앞서 구현한 naver.py의 NaverProvider 클래스를 사용하여 MCP 서버에 네이버 로그인 인증을 적용합니다. 클라이언트가 이 MCP 서버에 접근하면 자동으로 네이버 로그인 페이지로 리다이렉트되며, 사용자가 권한을 승인하면 서버를 사용할 수 있게 됩니다. 이번 예제에서 다루는 MCP 서버에는 총 세 가지의 도구를 등록합니다. get_current_time: 현재 시간을 반환하는 도구로, 정확한 시간 계산을 돕습니다. web_search: 네이버 검색을 수행하는 도구입니다. create_calendar_schedule: 네이버 캘린더에 일정을 추가하는 도구로, 해당 도구를 사용하기 위해서는 인증이 필요합니다. 다음은 네이버 인증이 적용된 MCP 서버 구현 코드입니다. # remote_auth_server.py import os import re import uuid import httpx from datetime import datetime from fastmcp import FastMCP from naver import NaverProvider # 커스터마이징한 naver.py from dotenv import load_dotenv load_dotenv() NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID") NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET") BASE_URL = os.getenv("BASE_URL") # 네이버 OAuth Provider 초기화 auth = NaverProvider( client_id=NAVER_CLIENT_ID, client_secret=NAVER_CLIENT_SECRET, base_url=BASE_URL, ) mcp = FastMCP(name="Naver Remote MCP", auth=auth) # Naver OAuth 인증 적용 # 현재 시간 도구 정의 @mcp.tool( name="get_current_time", description="현재 시간을 반환하는 도구입니다. 한국 시간(KST, UTC+9) 기준으로 현재 날짜와 시간을 제공합니다." ) # openAI API는 빈 properties를 허용하지 않으므로 더미 파라미터 설정 async def get_current_time(format: str = "default") -> dict: """ 현재 시간을 조회합니다. Args: format (str): 사용하지 않음 (호환성을 위한 더미 파라미터) Returns: { "datetime": str, # ISO 8601 형식 (YYYY-MM-DDTHH:MM:SS) "date": str, # 현재 날짜 (YYYY-MM-DD) "time": str, # 현재 시간 (HH:MM:SS) "weekday": str, # 요일 (한글) "timestamp": int # Unix timestamp } """ import pytz # 한국 시간대 설정 kst = pytz.timezone('Asia/Seoul') now = datetime.now(kst) weekdays = ['월요일', '화요일', '수요일', '목요일', '금요일', '토요일', '일요일'] return { "datetime": now.strftime("%Y-%m-%dT%H:%M:%S"), "date": now.strftime("%Y-%m-%d"), "time": now.strftime("%H:%M:%S"), "weekday": weekdays[now.weekday()], "timestamp": int(now.timestamp()) } # 네이버 검색 도구 정의 @mcp.tool( name="web_search", description="네이버 웹 검색을 수행하는 도구입니다." ) async def web_search(query: str, display: int = 10, start: int = 1, sort: str = "sim") -> dict: """ 네이버 웹 검색을 수행합니다. Args: query (str): 검색어. 사용자의 요청을 반영한 구체적이고 문맥이 드러나는 검색 질의(ex. 2025년 개봉 예정인 SF 영화). 시점, 조건, 문맥 등을 포함할 것. display (int): 출력 건수 (1~100) start (int): 검색 시작 위치 (1~1000) sort (str): 정렬 옵션 ("sim" | "date") Returns: { "query": str, "total": int, "items": [{"title": str, "link": str, "description": str}] } """ url = "https://openapi.naver.com/v1/search/webkr.json" params = {"query": query, "display": display, "start": start, "sort": sort} headers = { "X-Naver-Client-Id": NAVER_CLIENT_ID, "X-Naver-Client-Secret": NAVER_CLIENT_SECRET, } async with httpx.AsyncClient() as client: r = await client.get(url, headers=headers, params=params) r.raise_for_status() data = r.json() results = [] for item in data.get("items", []): results.append({ "title": re.sub(r"<.*?>", "", item.get("title") or "").strip(), "link": item.get("link"), "description": re.sub(r"<.*?>", "", item.get("description") or "").strip(), }) return { "query": query, "total": data.get("total", 0), "items": results, } # 네이버 캘린더 일정 추가 도구 정의 @mcp.tool( name="create_calendar_schedule", description="네이버 캘린더에 새로운 일정을 추가하는 도구입니다." ) async def create_calendar_schedule( title: str, start_datetime: str, end_datetime: str, description: str = "", location: str = "", ) -> dict: """ 캘린더에 일정을 추가합니다. Args: title (str): 일정의 제목을 추가합니다. start_datetime (str): 일정이 시작하는 시간을 설정합니다. end_datetime (str): 일정이 끝나는 시간을 설정합니다. description (str): 일정에 대한 설명을 추가합니다. location (str): 일정의 장소를 추가합니다. Returns: { "success": bool, "message": str, "response": str # (성공 시) } 또는 실패 시 { "success": bool, "error_code": int, # (예외 발생 시 "error": str) "message": str } """ from fastmcp.server.dependencies import get_access_token import urllib.request import urllib.parse from datetime import datetime as dt try: token = get_access_token() access_token = token.token header = "Bearer " + access_token url = "https://openapi.naver.com/calendar/createSchedule.json" # 날짜 형식 변환 네이버 캘린더가 지원하는 형식: "YYYYMMDDTHHMMSS" def parse_datetime(dt_str): # 이미 올바른 형식이면 그대로 반환 if 'T' in dt_str and len(dt_str) == 15: return dt_str # "YYYY-MM-DD HH:MM:SS" 형식 변환 try: parsed = dt.strptime(dt_str.replace('T', ' ').replace('-', '').replace(':', ''), '%Y%m%d %H%M%S') return parsed.strftime('%Y%m%dT%H%M%S') except: pass # "YYYY-MM-DD HH:MM:SS" 형식 try: parsed = dt.strptime(dt_str, '%Y-%m-%d %H:%M:%S') return parsed.strftime('%Y%m%dT%H%M%S') except: pass return dt_str start_datetime = parse_datetime(start_datetime) end_datetime = parse_datetime(end_datetime) # URL 인코딩 cal_sum = urllib.parse.quote(title) cal_des = urllib.parse.quote(description) if description else "" cal_loc = urllib.parse.quote(location) if location else "" # 고유 ID 생성 uid = str(uuid.uuid4())[:14] # iCalendar 문자열 생성 schedule_ical = "BEGIN:VCALENDAR\n" schedule_ical += "VERSION:2.0\n" schedule_ical += "PRODID:Naver Calendar\n" schedule_ical += "CALSCALE:GREGORIAN\n" schedule_ical += "BEGIN:VTIMEZONE\n" schedule_ical += "TZID:Asia/Seoul\n" schedule_ical += "BEGIN:STANDARD\n" schedule_ical += "DTSTART:19700101T000000\n" schedule_ical += "TZNAME:GMT%2B09:00\n" schedule_ical += "TZOFFSETFROM:%2B0900\n" schedule_ical += "TZOFFSETTO:%2B0900\n" schedule_ical += "END:STANDARD\n" schedule_ical += "END:VTIMEZONE\n" schedule_ical += "BEGIN:VEVENT\n" schedule_ical += "SEQUENCE:0\n" schedule_ical += "CLASS:PUBLIC\n" schedule_ical += "TRANSP:OPAQUE\n" schedule_ical += f"UID:{uid}\n" schedule_ical += f"DTSTART;TZID=Asia/Seoul:{start_datetime}\n" schedule_ical += f"DTEND;TZID=Asia/Seoul:{end_datetime}\n" schedule_ical += f"SUMMARY:{cal_sum} \n" if description: schedule_ical += f"DESCRIPTION:{cal_des} \n" if location: schedule_ical += f"LOCATION:{cal_loc} \n" now = datetime.now().strftime("%Y%m%dT%H%M%SZ") schedule_ical += f"CREATED:{now}\n" schedule_ical += f"LAST-MODIFIED:{now}\n" schedule_ical += f"DTSTAMP:{now}\n" schedule_ical += "END:VEVENT\n" schedule_ical += "END:VCALENDAR" data = "calendarId=defaultCalendarId&scheduleIcalString=" + schedule_ical request = urllib.request.Request(url, data=data.encode("utf-8")) request.add_header("Authorization", header) response = urllib.request.urlopen(request) rescode = response.getcode() if rescode == 200: response_body = response.read() result = response_body.decode('utf-8') return { "success": True, "message": "일정이 성공적으로 추가되었습니다.", "response": result } else: return { "success": False, "error_code": rescode, "message": f"일정 추가에 실패했습니다." } except Exception as e: return { "success": False, "error": str(e), "message": "일정 추가 중 오류가 발생했습니다." } if __name__ == "__main__": mcp.run(transport="streamable-http", path="/mcp") 5. FastMCP 기반 클라이언트 구현 본 프로젝트에서 FastMCP의 OAuth Proxy를 기반으로 한 NaverProvider를 사용하여 인증이 적용된 MCP 서버를 구성했습니다. FastMCP 클라이언트는 NaverProvider의 OAuth 인증 플로우(브라우저 로그인, PKCE, 토큰 관리 등) 를 완전히 지원하므로, 이 서버와의 통신을 위해서는 FastMCP 기반 클라이언트 구현이 필요합니다. 다음의 클라이언트 스크립트를 실행한 뒤 인증 과정을 거치면 '~/.fastmcp/oauth-mcp-client-cache' 경로에 토큰이 발급됩니다. 그리고 클라이언트는 해당 토큰을 가지고 MCP 서버에 접근할 수 있게 됩니다. 모델은 HCX-005를 사용하였고, 만약 HCX-007 모델을 도구와 함께 사용할 경우에는 reasoning_effort="none"을 반드시 설정해야 합니다. # fastmcp_client.py import os import json import glob import asyncio from typing import List from fastmcp import Client # https://gofastmcp.com/clients/client#the-fastmcp-client from langchain.agents import create_agent from langchain.messages import SystemMessage, HumanMessage, AIMessage from langchain_naver import ChatClovaX from langchain_core.tools import StructuredTool from pydantic import create_model from dotenv import load_dotenv load_dotenv() async def main(clova_api_key: str, server_url: str): """ FastMCP 클라이언트를 실행합니다. 인증 기반 MCP 서버와 연결하여 인증 절차를 진행하고 서버에 등록된 도구를 연동하여 최종 응답을 생성합니다. Args: clova_api_key: CLOVA Studio API Key. server_url: MCP 서버의 URL. """ model = ChatClovaX(model="HCX-005", api_key=clova_api_key) # FastMCP 클라이언트 생성 및 OAuth 인증 async with Client(server_url, auth="oauth") as auth_client: print("✓ Authenticated with OAuth!") # MCP 도구를 LangChain 형식으로 변환 structured_tools = await load_tools(auth_client) agent = create_agent(model, structured_tools) state = { "messages": [ SystemMessage(content=( "당신은 비서입니다. 사용자가 원하는 것을 도와주세요\n" "당신은 현재 시간을 알 수 있고, 웹 검색을 하거나 캘린더에 일정을 추가 할 수 있습니다\n" "사용자 요청에 대해 현재 시간이 필요한 경우에는 정확한 시간을 활용해 주세요\n" "예를 들어, 사용자가 '지금 몇 시야?' 라고 요청하면 현재 시간을 알려줍니다\n " "사용자가 '내일 일정 만들어줘','3일 뒤로 일정 잡아줘' 라고 요청하면 현재 시간을 기준으로 날짜를 확인해 활용해주세요\n" "각 도구는 반드시 한 번만 호출하세요. 일정 추가(create_calendar_schedule)는 절대 중복 호출하지 마세요\n" )) ] } print("\nAI: 안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.)") while True: user_input = input("\n\nUser: ") if user_input.lower() in ["종료", "exit"]: print("AI: 대화를 종료합니다. 이용해주셔서 감사합니다.") break state["messages"].append(HumanMessage(content=user_input)) # astream_events를 사용하여 스트리밍으로 결과 처리 try: final_answer = "" async for event in agent.astream_events(state, version="v1"): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: print(chunk.content, end="", flush=True) final_answer += chunk.content elif kind == "on_tool_start": print(f"\n[도구 선택]: {event['name']}\n[도구 호출]: {event['data'].get('input')}") elif kind == "on_tool_end": print(f"[도구 응답]: {event['data'].get('output')}\n") # 스트리밍이 끝나면 최종 답변을 AIMessage로 만들어 상태에 추가 state["messages"].append(AIMessage(content=final_answer)) except Exception as e: print(f"\nAI: 요청을 처리하는 중에 오류가 발생했습니다. 오류: {e}") pass async def load_tools(client: Client) -> List[StructuredTool]: """ FastMCP 클라이언트로부터 도구를 로드하여 LangChain StructuredTool로 변환합니다. Args: client: FastMCP Client 인스턴스 (인증된 세션 유지) Returns: List[StructuredTool]: LangChain 호환 도구 목록 """ # MCP 서버에서 도구 목록 가져오기 tools = await client.list_tools() structured_tools = [] for tool in tools: # 도구의 입력 스키마 추출 schema = tool.inputSchema or {} props = schema.get("properties", {}) # 파라미터 없는 도구 제외 if not props: continue # Pydantic 모델 동적 생성 - MCP 스키마를 Pydantic 필드로 변환 field_definitions = {} for key, prop in props.items(): # JSON Schema 타입을 Python 타입으로 변환 field_type = _get_python_type(prop.get("type", "string")) # 모든 필드를 required로 설정 (... = Ellipsis) field_definitions[key] = (field_type, ...) # 동적으로 Pydantic 모델 클래스 생성 # 예: "web_search" -> "web_searchInput" 클래스 InputModel = create_model( f"{tool.name}Input", **field_definitions ) # 각 도구마다 고유한 async 함수 생성 # tool_name과 mcp_client를 캡처하여 각 도구마다 고유한 함수 생성 def create_async_func(tool_name=tool.name, mcp_client=client): async def func(**kwargs): # 실제 MCP 서버의 도구 호출 result = await mcp_client.call_tool(tool_name, kwargs) return result return func # LangChain StructuredTool 생성 # coroutine 파라미터를 사용하여 async 함수 지원 structured_tool = StructuredTool.from_function( coroutine=create_async_func(), name=tool.name, description=tool.description or "", args_schema=InputModel, ) structured_tools.append(structured_tool) return structured_tools def _get_python_type(json_type: str) -> type: """JSON Schema 타입을 Python 타입으로 변환합니다.""" type_mapping = { "string": str, "integer": int, "number": float, "boolean": bool, "array": list, "object": dict, } return type_mapping.get(json_type, str) if __name__ == "__main__": CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") BASE_URL = os.getenv("BASE_URL") SERVER_URL = BASE_URL + "/mcp/" asyncio.run(main(CLOVA_STUDIO_API_KEY, SERVER_URL)) 6. 시나리오 실행 본 섹션에서는 위에서 구현한 클라이언트를 실행했을 때 나타나는 단계를 살펴봅니다. 네이버 로그인 및 권한 부여 1. FastMCP 클라이언트를 실행하면 다음과 같은 화면이 나타납니다. 2. Allow Access 버튼을 눌러주면, 네이버 로그인 페이지로 넘어가집니다. 3. 로그인을 완료하면 권한 부여 페이지가 나타납니다. 캘린더 일정 추가 도구를 사용하려면 캘린더 일정 등록 권한에 반드시 동의해야 합니다. 4. 권한 부여가 완료되면 FastMCP의 인증 완료 페이지로 리다이렉트됩니다. 대화 실행 결과 아래 예시는 사용자가 두 차례 질의를 이어가는 과정에서, 인증 기반 MCP 서버의 web_search 도구를 호출하여 장소를 검색하고, get_current_time 도구와 create_calendar_schedule 도구를 호출하여 해당 장소에 방문하는 일정을 추가합니다. 해당 과정에서 누적된 대화 히스토리가 활용되는 흐름을 보여줍니다. AI: 안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.) User: 예술의 전당이 어디있어? [도구 선택]: web_search [도구 호출]: {'query': '예술의 전당 위치', 'display': 1, 'start': 1, 'sort': 'date'} [도구 응답]: content='CallToolResult(content=[TextContent(type=\'text\', text=\'{"query":"예술의 전당 위치","total":3145637,"items":[{"title":"오시는 길·주차안내 | 방문·이용","link":"https://www.sac.or.kr/site/main/welcome/welcome?parkingInfo=y","description":"주소 도로명 : 서울시 서초구 남부순환로 2406 예술의전당 (우 06757), 지번 : 서울시 서초구 서초동 700번지, 건물명 : 예술의전당(오페라하우스, 음악당, 한가람미술관,한가람디자인미술관, 서울서예박물관)"}]}\', annotations=None, meta=None)], structured_content={\'query\': \'예술의 전당 위치\', \'total\': 3145637, \'items\': [{\'title\': \'오시는 길·주차안내 | 방문·이용\', \'link\': \'https://www.sac.or.kr/site/main/welcome/welcome?parkingInfo=y\', \'description\': \'주소 도로명 : 서울시 서초구 남부순환로 2406 예술의전당 (우 06757), 지번 : 서울시 서초구 서초동 700번지, 건물명 : 예술의전당(오페라하우스, 음악당, 한가람미술관,한가람디자인미술관, 서울서예박물관)\'}]}, data={\'query\': \'예술의 전당 위치\', \'total\': 3145637, \'items\': [{\'title\': \'오시는 길·주차안내 | 방문·이용\', \'link\': \'https://www.sac.or.kr/site/main/welcome/welcome?parkingInfo=y\', \'description\': \'주소 도로명 : 서울시 서초구 남부순환로 2406 예술의전당 (우 06757), 지번 : 서울시 서초구 서초동 700번지, 건물명 : 예술의전당(오페라하우스, 음악당, 한가람미술관,한가람디자인미술관, 서울서예박물관)\'}]}, is_error=False)' name='web_search' tool_call_id='call_DW7Ti30f1EeSZ5tXBZAVebsR' 예술의 전당은 **서울시 서초구 남부순환로 2406**에 위치해 있으며, 우편번호는 06757입니다. 건물명은 오페라하우스, 음악당, 한가람미술관, 한가람디자인미술관, 서울서예박물관이 포함되어 있습니다. 보다 자세한 정보와 주차 안내는 공식 웹사이트인 [https://www.sac.or.kr](https://www.sac.or.kr)에서 확인하실 수 있습니다. User: 3일 뒤 오후 3시에 방문 일정 잡아줘. [도구 선택]: get_current_time [도구 호출]: {'format': 'yyyy-MM-dd HH:mm'} [도구 응답]: content='CallToolResult(content=[TextContent(type=\'text\', text=\'{"datetime":"2025-10-15T14:48:06","date":"2025-10-15","time":"14:48:06","weekday":"수요일","timestamp":1760507286}\', annotations=None, meta=None)], structured_content={\'datetime\': \'2025-10-15T14:48:06\', \'date\': \'2025-10-15\', \'time\': \'14:48:06\', \'weekday\': \'수요일\', \'timestamp\': 1760507286}, data={\'datetime\': \'2025-10-15T14:48:06\', \'date\': \'2025-10-15\', \'time\': \'14:48:06\', \'weekday\': \'수요일\', \'timestamp\': 1760507286}, is_error=False)' name='get_current_time' tool_call_id='call_mLl81eYuX5SIsUINUOCFjJ2q' [도구 선택]: create_calendar_schedule [도구 호출]: {'title': '예술의 전당 방문', 'start_datetime': '2025-10-18 15:00', 'end_datetime': '2025-10-18 16:00', 'description': '예술의 전당 방문 일정', 'location': '서울시 서초구 남부순환로 2406'} [도구 응답]: content='CallToolResult(content=[TextContent(type=\'text\', text=\'{"success":true,"message":"일정이 성공적으로 추가되었습니다.","response":"{\\\\"result\\\\":\\\\"success\\\\",\\\\"code\\\\":200,\\\\"returnValue\\\\":{\\\\"calendarId\\\\":\\\\"5946777\\\\",\\\\"processType\\\\":\\\\"create\\\\",\\\\"icalUid\\\\":\\\\"f518004f-459f-\\\\"}}"}\', annotations=None, meta=None)], structured_content={\'success\': True, \'message\': \'일정이 성공적으로 추가되었습니다.\', \'response\': \'{"result":"success","code":200,"returnValue":{"calendarId":"5946777","processType":"create","icalUid":"f518004f-459f-"}}\'}, data={\'success\': True, \'message\': \'일정이 성공적으로 추가되었습니다.\', \'response\': \'{"result":"success","code":200,"returnValue":{"calendarId":"5946777","processType":"create","icalUid":"f518004f-459f-"}}\'}, is_error=False)' name='create_calendar_schedule' tool_call_id='call_CfxP1vBw9mmGkuheiM1KzsJb' 네, 3일 뒤인 **2025년 10월 18일 오후 3시**부터 1시간 동안 예술의 전당 방문 일정이 잡혔습니다. 해당 일정은 사용자님의 캘린더에 저장되었으며, 주소는 서울시 서초구 남부순환로 2406입니다. 즐거운 관람 되세요! 마무리 3부에서는 FastMCP에서 제공하는 인증 시스템을 MCP 서버에 통합하는 과정을 살펴보았습니다. NaverProvider를 예시로 커스터마이징 한 인증 클래스를 구현하고, FastMCP의 OAuth Proxy를 통해 인증 과정을 자동화함으로써, HyperCLOVA X 모델과 안전하게 연동되는 개인화 MCP 서버를 완성했습니다. 마지막 4부에서는 운영과 확장에 대한 팁을 다룹니다. MCP Inspector를 활용한 서버 디버깅, Flowise를 이용한 시각적 테스트, 그리고 직접 서버를 작성하고 확장하는 실전 노하우를 소개합니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 링크 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 링크

감사 합니다. 문제가 해결 되었네요

들어가며 (1부) MCP 실전 쿡북: LangChain에서 네이버 검색 도구 연결하기에 이어서, 이번 2부에서는 MCP로 Stateful 대화 기능을 구현하고 흐름을 관리하는 방법을 다룹니다. 1부에서 만든 MCP 서버를 기반으로 LangChain과 HyperCLOVA X 모델을 연결하고, 세션 ID를 활용해 대화 흐름을 지속적으로 이어가는 과정을 살펴봅니다. 또한 LangGraph의 Checkpointer를 사용해 세션 상태를 저장·재사용하며, SQLite 기반 저장소를 통해 멀티턴 문맥을 자연스럽게 유지하는 방법을 소개합니다. 마지막으로, 제한된 토큰 수 내에서 모델의 컨텍스트를 효율적으로 관리하는 팁도 함께 다뤄보겠습니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 링크 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 링크 1. 사전 준비 사항 이 예제를 실행하려면 Python 3.10 이상의 개발 환경이 필요합니다. 가상환경을 구성하고 .env 파일에 API 키를 등록한 뒤, 필요한 패키지를 설치합니다. MCP 서버(server.py)는 1부에서 생성한 서버 파일을 그대로 사용합니다. 프로젝트 구성 프로젝트의 전체 파일 구조는 다음과 같습니다. Python 버전은 3.10 이상, 3.13 미만입니다. mcp_cookbook_part2/ ├── server.py # mcp_cookbook_part1에서 생성한 파일을 사용합니다. ├── client/ │ ├── stateful_client.py │ ├── stateful_client_trim.py │ └── stateful_client_sum.py ├── init_db.py ├── checkpoint.db # init_db.py 실행 시 생성됩니다. ├── requirements.txt ├── .venv └── .env 환경 변수 설정 루트 디렉터리에 .env 파일을 생성한 뒤, 앞서 발급받은 API Key를 다음과 같이 입력하고 저장합니다. 이때 따옴표 없이 값을 작성해야 하며, VS Code에서 실행할 경우 설정에서 Use Env File 옵션이 활성화되어 있는지 확인하세요. CLOVA_STUDIO_API_KEY=YOUR_API_KEY NAVER_CLIENT_ID=YOUR_CLIENT_ID NAVER_CLIENT_SECRET=YOUR_CLIENT_SECRET 패키지 설치 프로젝트에 필요한 패키지 목록은 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 루트 디렉터리에 requirements.txt 파일로 저장하세요. requirements.txt 다운로드 루트 디렉터리에서 터미널을 실행하여 다음과 같이 예제 실행에 필요한 패키지를 설치합니다. 가상환경 설치를 권장합니다. # 1. 파이썬 가상환경 생성 python -m venv .venv # 2. 가상환경 활성화 (macOS/Linux) source .venv/bin/activate # (Windows) # .venv/Scripts/activate.ps1 # 3. 패키지 설치 pip install -r requirements.txt 2. Stateless와 Stateful 개념 대화 시스템은 크게 Stateless와 Stateful 두 가지 방식으로 동작할 수 있습니다. Stateless: 각 요청이 이전 맥락과 완전히 분리되어 독립적으로 처리됩니다. Stateful: 대화 상태를 별도의 저장소에 보관하고, 동일한 세션 ID를 가진 후속 요청에서 자동으로 이 상태를 불러옵니다. 이렇게 하면 모델이 이전 대화 내용을 자연스럽게 이어받아 연속적이고 일관된 대화 흐름을 유지할 수 있습니다. 장기 대화나 도구 호출이 섞이는 복합 시나리오에서는 Stateful 방식을 사용하는 것이 사용자의 의도와 맥락을 안정적으로 추적하는 데 도움이 됩니다. 아래 도식은 클라이언트가 사용자 입력을 받아 세션 저장소에서 상태를 조회하거나 초기 컨텍스트를 구성한 뒤, LLM과 상호작용하여 응답하고, 마지막에 상태를 저장하는 Stateful 대화 흐름을 단계별로 보여줍니다. 3. Stateful 대화 구현 앞서 제시한 도식 흐름을 실제 코드로 구현하는 방법을 설명합니다. 세션 저장소 초기화 LangGraph에서 대화 상태를 지속적으로 저장・조회하려면 세션 저장소가 필요합니다. 여기에서는 SQLAlchemy를 이용해 SQLite 데이터베이스 파일(checkpoint.db)을 생성하고, 이후 Checkpointer가 사용할 수 있는 기본 테이블 메타데이터를 초기화합니다. 아래 스크립트를 실행하면 checkpoint.db 파일이 생성되며, 최초 1회만 실행하면 됩니다. 그 이후부터는 이 데이터베이스가 세션별 대화 상태를 보관하는 저장소로 사용됩니다. from sqlalchemy import create_engine from sqlalchemy.orm import declarative_base, sessionmaker engine = create_engine("sqlite:///checkpoint.db", connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) Base = declarative_base() Base.metadata.create_all(bind=engine) Stateful 대화 실행 다음은 사용자가 입력한 세션 ID를 기준으로 대화 상태를 관리하는 클라이언트입니다. 특정 ID가 입력되면 해당 세션의 히스토리를 불러와 맥락을 이어가고, 새로운 ID를 입력하면 시스템 프롬프트와 함께 새로운 세션을 시작합니다. 각 턴이 종료될 때마다 LangGraph가 Checkpointer에 상태를 저장하기 때문에, 이후 세션을 재실행하면 직전까지의 대화 흐름을 그대로 이어받아 자연스럽고 일관된 멀티턴 대화를 이어갈 수 있습니다. 모델은 HCX-005를 사용하였고, 만약 HCX-007 모델을 도구와 함께 사용할 경우에는 reasoning_effort="none"을 반드시 설정해야 합니다. import os import asyncio import uuid from dotenv import load_dotenv from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain_naver import ChatClovaX from langchain.agents import create_agent from langchain.messages import SystemMessage, HumanMessage from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver async def main(clova_api_key: str, server_url: str, checkpoint_path: str): """ CLOVA Studio 모델을 LangChain을 통해 호출하고, MCP 서버에 등록된 도구를 연동하여 최종 응답을 생성합니다. 사용자 입력에 따라 새 세션을 시작하거나 이전 세션을 이어갑니다. Args: clova_api_key (str): CLOVA Studio에서 발급받은 API Key server_url (str): 연결할 MCP 서버의 엔드포인트 URL checkpoint_path (str): Checkpoint 저장 경로 """ model = ChatClovaX(model="HCX-005", api_key=clova_api_key) async with streamablehttp_client(server_url) as (read, write, _,): async with ClientSession(read, write) as session: # MCP 서버를 초기화하고 도구 목록을 불러옵니다. await session.initialize() tools = await load_mcp_tools(session) # Checkpointer를 생성합니다. async with AsyncSqliteSaver.from_conn_string(checkpoint_path) as checkpointer: # 에이전트를 생성하고 Checkpointer를 연결합니다. agent = create_agent(model, tools, checkpointer=checkpointer) # 참조할 세션 ID를 입력받습니다. thread_id = input("세션 ID를 입력하세요(새 세션 시작은 Enter):").strip() # 새 세션이면 신규 세션 ID를 생성합니다. if not thread_id: thread_id = str(uuid.uuid4()) config = {"configurable": {"thread_id": thread_id}} print(f"현재 세션 ID: {thread_id}\n") print("안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.)") system_message = [ SystemMessage(content=( "당신은 친절한 AI 어시스턴트입니다." "사용자의 질문에 대해 신뢰할 수 있는 정보만 근거로 삼아 답변하세요." )) ] while True: user_input = input("\n\nUser: ") if user_input.lower() in ["종료", "exit"]: print("대화를 종료합니다. 이용해 주셔서 감사합니다.") break current_state = {"messages": [HumanMessage(content=user_input)]} try: existing_checkpoint = await checkpointer.aget_tuple(config) if existing_checkpoint is not None: # 기존 스레드가 있으면 현재 메시지만 추가합니다. state = current_state else: # 새 스레드라면 SystemMessage와 현재 메시지를 모두 추가합니다. state = {"messages": system_message + current_state["messages"]} # astream_events를 사용하여 스트리밍으로 응답을 처리합니다. async for event in agent.astream_events(state, config=config, version="v1"): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: print(chunk.content, end="", flush=True) elif kind == "on_tool_start": print(f"\n[도구 선택]: {event['name']}\n[도구 호출]: {event['data'].get('input')}") elif kind == "on_tool_end": print(f"[도구 응답]: {event['data'].get('output')}\n") except Exception as e: print(f"\n요청을 처리하는 중에 오류가 발생했습니다. 오류: {e}") pass if __name__ == "__main__": """ .env 파일에서 CLOVA Studio API Key를 로드하고, MCP 서버의 엔드포인트 URL을 설정한 후 클라이언트를 실행합니다. """ load_dotenv() CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") SERVER_URL = "http://127.0.0.1:8000/mcp/" CHECKPOINT_PATH = "checkpoint.db" asyncio.run(main(CLOVA_STUDIO_API_KEY, SERVER_URL, CHECKPOINT_PATH)) 위 스크립트를 실행하면, CLOVA Studio 모델(HCX-005)이 MCP 서버를 통해 등록된 도구와 연동되어 사용자의 질문에 응답합니다. 특정 세션에서 멀티턴 대화를 이어가면 대화 상태가 자동으로 저장되며, 이후 세션을 종료했다가 다시 실행하더라도 직전까지의 맥락을 불러와 자연스럽게 대화를 지속할 수 있습니다. 또한 이러한 대화 기록은 checkpoint.db의 checkpoints 테이블 내 checkpoint 컬럼에 직렬화된 형태로 저장되며, 실제 파일을 열어보면 해당 값에서 세션별 상태를 직접 확인할 수 있습니다. 다음과 같이 새로운 세션을 시작하면, 기존에 저장된 대화 맥락은 참조되지 않으며 시스템 프롬프트와 함께 처음부터 대화를 이어가게 됩니다. 다음과 같이 기존 세션을 재시작하면, 직전까지 저장된 대화 맥락을 불러와 이전 흐름을 그대로 이어갈 수 있습니다. 4. 컨텍스트 관리 전략 Stateful 대화는 맥락을 자동으로 이어받는 장점이 있지만, 저장된 히스토리를 모두 모델 입력에 넣다 보면 토큰 수 초과 오류를 마주하게 됩니다. 이때 맥락 관리 및 토큰 최적화 전략이 중요합니다. 다음 전략은 OpenAI Cookbook: Context Engineering를 참고하였습니다. 서비스 성격과 요구 사항에 맞게 전략을 적절히 선택하거나 새롭게 조합해 보시길 바랍니다. 컨텍스트 트리밍(Context Trimming) 최근 N개 턴의 메시지만 유지하는 방식으로, 구현이 단순하고 추가 지연이 발생하지 않는다는 장점이 있습니다. 하지만 N개 이전의 맥락은 모두 잊히며, 최근 턴이라 하더라도 도구 호출 결과가 지나치게 길면 여전히 토큰 수 초과 오류가 발생할 수 있습니다. 여기에서 한 턴(Turn)은 사용자 메시지 1개와 그에 대한 처리 과정(도구 호출, 도구 응답, 최종 답변)을 모두 포함합니다. 다음 이미지는 컨텍스트 트리밍의 동작을 설명합니다. 예를 들어 N=2로 설정하면, 매번 새로운 요청이 들어올 때마다 오래된 대화가 순차적으로 제거되어, 모델은 항상 최근 2개의 턴만 참조하게 됩니다. 다음은 세션별로 턴의 개수를 카운트해 관리하는 예시 코드입니다. import os import asyncio import uuid from dotenv import load_dotenv from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain_naver import ChatClovaX from langchain.agents import create_agent from langchain.messages import SystemMessage, HumanMessage from langchain_core.messages import BaseMessage from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver MAX_TURNS_TO_KEEP = 5 # 최대 유지할 턴(Turn)의 개수 def trim_messages_by_turn(messages: list[BaseMessage], max_turns: int) -> list[BaseMessage]: """ 대화 기록을 턴(Turn) 단위로 트리밍합니다. SystemMessage를 제외한 최대 max_turns 개의 턴만 유지합니다. 턴은 HumanMessage로 시작하는 묶음으로 간주합니다. Args: messages (list[BaseMessage]): 대화 기록 메시지 리스트 max_turns (int): 유지할 최대 턴 수 """ if not messages: return [] # SystemMessage와 대화 기록을 분리합니다. system_message = messages[0] if isinstance(messages[0], SystemMessage) else None history_messages = messages[1:] if system_message else messages # 턴 시작 인덱스를 기준으로 현재 턴 개수를 계산합니다. turn_start_indices = [ i for i, msg in enumerate(history_messages) if isinstance(msg, HumanMessage) ] current_turn_count = len(turn_start_indices) if current_turn_count <= max_turns: # 현재 턴 개수가 최대 유지할 턴 수 이하이면 그대로 반환합니다. return messages # 유지할 턴의 시작 인덱스를 계산합니다. trim_index = current_turn_count - max_turns keep_messages_start_index = turn_start_indices[trim_index] # 메시지 리스트를 트리밍합니다. trimmed_history = history_messages[keep_messages_start_index:] print(f"\n{max_turns}턴을 초과하여서 트리밍을 수행합니다.\n") # SystemMessage와 트리밍된 대화 기록을 합쳐 반환합니다. return [system_message] + trimmed_history async def main(clova_api_key: str, server_url: str, checkpoint_path: str): """ CLOVA Studio 모델을 LangChain을 통해 호출하고, MCP 서버에 등록된 도구를 연동하여 최종 응답을 생성합니다. 사용자 입력에 따라 새 세션을 시작하거나 이전 세션을 이어갑니다. Args: clova_api_key (str): CLOVA Studio에서 발급받은 API Key server_url (str): 연결할 MCP 서버의 엔드포인트 URL checkpoint_path (str): Checkpoint 저장 경로 """ model = ChatClovaX(model="HCX-005", api_key=clova_api_key) async with streamablehttp_client(server_url) as (read, write, _,): async with ClientSession(read, write) as session: # MCP 서버를 초기화하고 도구 목록을 불러옵니다. await session.initialize() tools = await load_mcp_tools(session) # Checkpointer를 생성합니다. async with AsyncSqliteSaver.from_conn_string(checkpoint_path) as checkpointer: # 에이전트를 생성하고 Checkpointer를 연결합니다. agent = create_agent(model, tools, checkpointer=checkpointer) # 참조할 세션 ID를 입력받습니다. thread_id = input("세션 ID를 입력하세요(새 세션 시작은 Enter):").strip() # 새 세션이면 신규 세션 ID를 생성합니다. if not thread_id: thread_id = str(uuid.uuid4()) config = {"configurable": {"thread_id": thread_id}} print(f"현재 세션 ID: {thread_id}\n") print("안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.)") system_message = [ SystemMessage(content=( "당신은 친절한 AI 어시스턴트입니다." "사용자의 질문에 대해 신뢰할 수 있는 정보만 근거로 삼아 답변하세요." )) ] while True: user_input = input("\n\nUser: ") if user_input.lower() in ["종료", "exit"]: print("대화를 종료합니다. 이용해 주셔서 감사합니다.") break current_state = {"messages": [HumanMessage(content=user_input)]} try: existing_checkpoint = await checkpointer.aget_tuple(config) if existing_checkpoint is not None: # 기존 대화 기록을 불러옵니다. messages = existing_checkpoint.checkpoint["channel_values"]["messages"] # 대화 기록을 턴 단위로 트리밍합니다. trimmed_messages = trim_messages_by_turn(messages, MAX_TURNS_TO_KEEP) # 트리밍된 메시지에 현재 사용자 메시지를 추가합니다. state = {"messages": trimmed_messages + current_state["messages"]} else: # 새 스레드라면 SystemMessage와 현재 메시지를 모두 추가합니다. state = {"messages": system_message + [HumanMessage(content=user_input)]} # astream_events를 사용하여 스트리밍으로 응답을 처리합니다. async for event in agent.astream_events(state, config=config, version="v1"): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: print(chunk.content, end="", flush=True) elif kind == "on_tool_start": print(f"\n[도구 선택]: {event['name']}\n[도구 호출]: {event['data'].get('input')}") elif kind == "on_tool_end": print(f"[도구 응답]: {event['data'].get('output')}\n") except Exception as e: print(f"\n요청을 처리하는 중에 오류가 발생했습니다. 오류: {e}") pass if __name__ == "__main__": """ .env 파일에서 CLOVA Studio API Key를 로드하고, MCP 서버의 엔드포인트 URL을 설정한 후 클라이언트를 실행합니다. """ load_dotenv() CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") SERVER_URL = "http://127.0.0.1:8000/mcp/" CHECKPOINT_PATH = "checkpoint.db" asyncio.run(main(CLOVA_STUDIO_API_KEY, SERVER_URL, CHECKPOINT_PATH)) 위 스크립트를 실행하면, CLOVA Studio 모델(HCX-005)이 MCP 서버를 통해 등록된 도구와 연동되어 사용자의 질문에 응답합니다. 세션 단위로 대화 상태가 유지되므로 멀티턴 대화를 지속할 수 있으며, 설정한 최대 턴 수를 기준으로 컨텍스트 트리밍이 적용됩니다. 다음은 사용자가 질의를 이어갈 때 트리밍이 발생하는 시점을 보여줍니다. 예를 들어, 최대 턴 수를 5로 설정한 경우, 여섯 번째 요청까지는 직전 5개 턴이 모두 유지되어 트리밍이 발생하지 않습니다. 일곱 번째 요청부터는 가장 오래된 턴이 제거되어, 항상 최근 5개 턴만 참조하도록 트리밍이 수행됩니다. 컨텍스트 요약(Context Summarization) 모델을 이용해 이전 대화를 간결한 요약으로 변환하고, 해당 요약 메시지를 히스토리에 삽입하는 방식입니다. 이 방법은 장기 기억을 압축해 보존할 수 있다는 장점이 있지만, 요약 과정에서 세부 정보가 누락될 수 있고, 요약을 갱신할 때마다 모델 호출이 필요하다는 유의 사항이 있습니다. 요약이 실행되는 로직은 다음과 같습니다. 여기에서 한 턴(Turn)은 사용자 메시지 1개와 그에 대한 처리 과정(도구 호출, 도구 응답, 최종 답변)을 모두 포함합니다. CONTEXT_TURNS_LIMIT(N): 요약을 실행하는 기준이 되는 값입니다. 누적된 턴의 개수가 이 값보다 클 때 요약을 실행합니다. MAX_TURNS_TO_KEEP(M): 요약이 실행된 뒤에도 원문 그대로 유지할 최근 턴 개수입니다. 그 이전 구간은 요약문으로 대체됩니다. 즉, 누적된 턴의 개수가 N을 초과하면 요약이 실행되고, 그 시점에서 최근 M개의 턴만 상태에 유지하며 그 이전 대화는 두 개의 메시지(요약 요청·요약 결과)로 치환됩니다. HumanMessage: 지금까지의 대화를 요약해 주세요. AIMessage: {요약문} 다음 이미지는 컨텍스트 요약이 동작하는 기준과 흐름을 설명합니다. 요약 기준 턴 수(Max N)를 초과하면 모델은 오래된 대화 중 요약 대상 구간(N − M + 1개의 턴)을 압축해 하나의 요약문으로 변환합니다. 이때 요약 요청(HumanMessage)과 요약 결과(AIMessage) 두 메시지가 새로 삽입되고, 그 뒤로는 보존 대상 구간(최근 M개의 사용자 턴)이 원문 그대로 유지됩니다. 이때, 요약이 요청되는 시점에 새로 들어오는 사용자 입력이 함께 포함되므로, 실제 요약 대상 범위는 N − M + 1개의 턴으로 계산됩니다. 다음은 세션별로 컨텍스트를 요약해 관리하는 예시 코드입니다. import os import asyncio import uuid from dotenv import load_dotenv from typing import List, Optional, Tuple from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain_naver import ChatClovaX from langchain.agents import create_agent from langchain.messages import SystemMessage, HumanMessage, AIMessage from langchain_core.messages import BaseMessage from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver CONTEXT_TURNS_LIMIT = 5 # 요약을 시작할 기준이 되는 턴(Turn)의 개수 MAX_TURNS_TO_KEEP = 2 # 최대 유지할 턴(Turn)의 개수 SUMMARY_REQUEST_CONTENT = "지금까지의 대화를 요약해 주세요." SUMMARY_PROMPT = ( "당신은 친절한 AI 어시스턴트입니다. 지금까지의 대화를 간결하게 요약하세요.\n" "요약에는 중요한 정보, 사용자 요청, 도구 호출 내역 단계가 포함되어야 합니다.\n" ) async def summarize_messages_by_turn(messages: List[BaseMessage], max_turns: int, context_limit: int, model: ChatClovaX, summary_prompt: str = SUMMARY_PROMPT,) -> Tuple[List[BaseMessage], bool]: # 반환 타입: (메시지 리스트, 요약 발생 여부) """ 대화 기록을 턴(Turn) 단위로 요약합니다. SystemMessage를 제외한 최대 max_turns 개의 턴만 유지합니다. 턴은 HumanMessage로 시작하는 묶음으로 간주합니다. 단, 요약은 현재 대화 기록의 사용자 턴 수가 context_limit를 초과할 때만 수행합니다. Args: messages (List[BaseMessage]): 대화 기록 메시지 리스트 max_turns (int): 유지할 최대 턴 수 context_limit (int): 요약을 시작할 기준이 되는 턴 수 model (ChatClovaX): 요약에 사용할 LLM 모델 summary_prompt (str): 요약에 사용할 프롬프트 """ if not messages: return [], False # SystemMessage와 대화 기록을 분리합니다. system_message: Optional[BaseMessage] = ( messages[0] if isinstance(messages[0], SystemMessage) else None ) # SystemMessage는 복사본을 만들어 메타데이터를 초기화합니다. if system_message: system_message = system_message.model_copy(deep=True) system_message.response_metadata = {} history_messages = messages[1:] if system_message else messages # 마지막 요약 지점을 찾습니다. last_synthetic_idx = -1 for idx, msg in enumerate(history_messages): if isinstance(msg, HumanMessage) and ( msg.additional_kwargs.get("synthetic") or getattr(msg, "content", "").strip() == SUMMARY_REQUEST_CONTENT ): last_synthetic_idx = idx # 요약 대상 범위를 결정합니다. if last_synthetic_idx != -1: target_messages = history_messages[last_synthetic_idx + 2 :] # 요청+응답 2개 이후 else: target_messages = history_messages # 실제 사용자 턴 개수를 계산합니다. turn_start_indices = [] for idx, msg in enumerate(target_messages): if isinstance(msg, HumanMessage) and not msg.additional_kwargs.get("synthetic"): turn_start_indices.append(idx) current_turn_count = len(turn_start_indices) # 요약 필요하지 않으면 원본 메시지를 그대로 반환합니다. if current_turn_count <= context_limit: return messages, False # 요약 경계 인덱스를 계산합니다. trim_index = current_turn_count - max_turns summary_boundary_in_target = turn_start_indices[trim_index] # 요약 대상 메시지를 준비합니다. summarize_messages_for_input = target_messages[:summary_boundary_in_target] conversation_snippets = [] if last_synthetic_idx != -1: old_synthetic_summary = history_messages[last_synthetic_idx + 1] prev_summary_text = getattr(old_synthetic_summary, "content", "").strip() if prev_summary_text: conversation_snippets.append( f"Previous Context Summary: {prev_summary_text}" ) for m in summarize_messages_for_input: content = (getattr(m, "content", "") or "").strip() if not content: continue role_label = "User" if isinstance(m, HumanMessage) else "Assistant" conversation_snippets.append(f"{role_label}: {content}") if not conversation_snippets: return messages, False summary_input_messages = [ SystemMessage(content=summary_prompt), HumanMessage(content="\n\n".join(conversation_snippets)), ] try: summary_response: BaseMessage = await model.ainvoke(summary_input_messages) summary_text = getattr(summary_response, "content", "").strip() except Exception as e: print(f"\n요약 호출 중 오류가 발생했습니다. 전체 기록을 유지합니다. 오류: {e}\n") return messages, False # 요약본과 함께 새로운 대화 상태를 구성합니다. shadow_user = HumanMessage( content=SUMMARY_REQUEST_CONTENT, additional_kwargs={"synthetic": True} ) synthetic_summary = AIMessage( content=summary_text, additional_kwargs={"synthetic": True} ) suffix_messages = target_messages[summary_boundary_in_target:] cleaned_suffix_messages = [] for msg in suffix_messages: new_msg = msg.model_copy(deep=True) new_msg.response_metadata = {} cleaned_suffix_messages.append(new_msg) new_history = [shadow_user, synthetic_summary] + cleaned_suffix_messages final_messages = ([system_message] + new_history) if system_message else new_history return final_messages, True def prune_to_last_summary(messages: List[BaseMessage]) -> List[BaseMessage]: """ Checkpointer에서 hydrate한 전체 메시지 중 마지막 요약 이후만 남겨 과거 히스토리 재유입을 차단합니다. Args: messages (List[BaseMessage]): 전체 메시지 리스트 """ if not messages: return messages system = messages[0] if isinstance(messages[0], SystemMessage) else None history = messages[1:] if system else messages base = 1 if system else 0 last_pair_start = -1 for i in range(len(history) - 1): h = history[i] a = history[i + 1] if isinstance(h, HumanMessage) and isinstance(a, AIMessage): if ( h.additional_kwargs.get("synthetic") or (getattr(h, "content", "").strip() == SUMMARY_REQUEST_CONTENT) ) and a.additional_kwargs.get("synthetic"): last_pair_start = i if last_pair_start == -1: return messages cut_pos = base + last_pair_start pruned = ([system] if system else []) + messages[cut_pos:] return pruned async def main(clova_api_key: str, server_url: str, checkpoint_path: str): """ CLOVA Studio 모델을 LangChain을 통해 호출하고, MCP 서버에 등록된 도구를 연동하여 최종 응답을 생성합니다. 사용자 입력에 따라 새 세션을 시작하거나 이전 세션을 이어갑니다. Args: clova_api_key (str): CLOVA Studio에서 발급받은 API Key server_url (str): 연결할 MCP 서버의 엔드포인트 URL checkpoint_path (str): Checkpoint 저장 경로 """ model = ChatClovaX(model="HCX-005", api_key=clova_api_key) async with streamablehttp_client(server_url) as (read, write, _,): async with ClientSession(read, write) as session: # MCP 서버를 초기화하고 도구 목록을 불러옵니다. await session.initialize() tools = await load_mcp_tools(session) # Checkpointer를 생성합니다. async with AsyncSqliteSaver.from_conn_string(checkpoint_path) as checkpointer: # 에이전트를 생성하고 Checkpointer를 연결합니다. agent = create_agent(model, tools, checkpointer=checkpointer) # 참조할 세션 ID를 입력받습니다. thread_id = input("세션 ID를 입력하세요(새 세션 시작은 Enter):").strip() # 새 세션이면 신규 세션 ID를 생성합니다. if not thread_id: thread_id = str(uuid.uuid4()) config = {"configurable": {"thread_id": thread_id}} print(f"현재 세션 ID: {thread_id}\n") print( "안녕하세요. 저는 AI 어시스턴트입니다. 원하시는 요청을 입력해 주세요. (종료하려면 '종료'를 입력하세요.)" ) system_message = [ SystemMessage(content=( "당신은 친절한 AI 어시스턴트입니다." "사용자의 질문에 대해 신뢰할 수 있는 정보만 근거로 삼아 답변하세요." )) ] while True: user_input = input("\n\nUser: ") if user_input.lower() in ["종료", "exit"]: print("대화를 종료합니다. 이용해 주셔서 감사합니다.") break current_state = {"messages": [HumanMessage(content=user_input)]} try: existing_checkpoint = await checkpointer.aget_tuple(config) if existing_checkpoint is not None: # 기존 대화 기록을 불러옵니다. messages: List[BaseMessage] = existing_checkpoint.checkpoint["channel_values"]["messages"] # 과거 히스토리 재유입 차단을 위해 마지막 요약 이후만 남깁니다. messages = prune_to_last_summary(messages) # 턴 단위로 컨텍스트 요약을 적용합니다. summarized_messages, is_summarized = await summarize_messages_by_turn( messages, MAX_TURNS_TO_KEEP, CONTEXT_TURNS_LIMIT, model ) if is_summarized: # 요약이 발생하면 요약본과 함께 현재 사용자 메시지를 추가합니다. state = {"messages": summarized_messages + current_state["messages"]} else: # 요약이 없으면 프루닝된 메시지에 현재 사용자 메시지만 추가합니다. state = {"messages": messages + current_state["messages"]} else: # 새 스레드라면 SystemMessage와 재 메시지를 모두 추가합니다. state = {"messages": system_message + [HumanMessage(content=user_input)]} # print(f"[state 확인] {state['messages']}\n") # astream_events를 사용하여 스트리밍으로 응답을 처리합니다. async for event in agent.astream_events(state, config=config, version="v1"): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: print(chunk.content, end="", flush=True) elif kind == "on_tool_start": print( f"\n[도구 선택]: {event['name']}\n[도구 호출]: {event['data'].get('input')}" ) elif kind == "on_tool_end": print(f"[도구 응답]: {event['data'].get('output')}\n") except Exception as e: print(f"\n요청을 처리하는 중에 오류가 발생했습니다. 오류: {e}") pass if __name__ == "__main__": """ .env 파일에서 CLOVA Studio API Key를 로드하고, MCP 서버의 엔드포인트 URL을 설정한 후 클라이언트를 실행합니다. """ load_dotenv() CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY") SERVER_URL = "http://127.0.0.1:8000/mcp/" CHECKPOINT_PATH = "checkpoint.db" asyncio.run(main(CLOVA_STUDIO_API_KEY, SERVER_URL, CHECKPOINT_PATH)) 위 스크립트를 실행하면, CLOVA Studio 모델(HCX-005)이 MCP 서버를 통해 등록된 도구와 연동되어 사용자의 질문에 응답합니다. 세션 단위로 대화 상태가 유지되므로 멀티턴 대화를 지속할 수 있으며, 설정된 요약 실행 임계치(N)와 최대 턴 수(M)를 기준으로 컨텍스트 요약이 자동 적용됩니다. 다음은 누적된 대화 상태를 보여줍니다. 예를 들어, N=5, M=2로 설정한 경우 여섯 번째 요청까지는 이전 5개 턴이 모두 유지되어 요약이 실행되지 않습니다. 일곱 번째 요청부터는 컨텍스트 요약이 실행되어, 이전 대화의 맥락이 두 개의 메시지(요약 요청·요약 결과)로 대체되고, 최근 두 개의 사용자 턴만 원문으로 유지됩니다. 마무리 2부에서는 세션 기반의 Stateful 대화 흐름을 구현하는 과정을 살펴보았습니다. 대화 상태를 저장하고 이어받는 방식, 그리고 컨텍스트를 효율적으로 관리하기 위한 전략을 실제 코드와 함께 구현해 보았는데요. 이를 통해 MCP와 HyperCLOVA X 모델을 함께 활용하면 대화의 맥락을 지속적으로 이해하고 기억하는 대화형 시스템을 설계할 수 있음을 확인했습니다. 다음 3부에서는 인증 구조를 통합하여 HyperCLOVA X 모델과 안전하게 연동되는 개인화 MCP 서버를 구현해 보겠습니다. MCP 실전 쿡북 4부작 시리즈 ✔︎ (1부) LangChain에서 네이버 검색 도구 연결하기 링크 ✔︎ (2부) 세션 관리와 컨텍스트 유지 전략 ✔︎ (3부) OAuth 인증이 통합된 MCP 서버 구축하기 링크 ✔︎ (4부) 운영과 확장을 위한 다양한 팁 링크

콘솔 > Maps > Application > Reverse Geocoding 체크하셨는지 확인 부탁드립니다. 설정했음에도 호출 에러가 발생한다면 공식 문의 채널로 접수 부탁드립니다. (해당 Client ID 로그 확인이 필요합니다.) - https://www.ncloud.com/support/question/service

인증키, 앱 패키지 이름 네이버 버전 해볼건 모두 해봤는데 안됩니다. xml에 여러가지 키 넣는방법도 1개씩 모두 따로도 해 봤는데 안됩니다. 어떤게 문제일까요..

저도 서버에서 maps - reverse geocoding api 호출하는데 동일한 문제 발생해서 댓글 남깁니다

사용가능합니다. 정상적으로 렌더링되지 않는다면 재현 코드 알려주시면 확인해보겠습니다.

strokeStyle: 'solid', 로만 사용합니다

혹시 어떤 API 사용하시는지 알 수 있을까요?

무슨 이유인지 모르겠으나 Application을 삭제후 다시 설정하고 docker 리빌드로 해결했습니다.

【오류 내용】 curl 'https://maps.apigw.ntruss.com/map-geocode/v2/geocode?query=강남구' \ -H 'x-ncp-apigw-api-key-id: 51fbsflpoh' \ -H 'x-ncp-apigw-api-key: 50MbjxRSImperC6WWRphD15bzHHOVJ1v7fpUHn3e' 응답: { "error": { "errorCode": "200", "message": "Authentication Failed", "details": "Invalid authentication information." } } 【질문】 1. Application 등록 외에 추가 설정이 필요한가요? 2. IAM 인증 키가 별도로 필요한가요? 3. VPC 환경에서 특별한 네트워크 설정이 필요한가요? 모든 설정이 공식 문서와 일치하는데도 401 오류가 계속됩니다. 도움 부탁드립니다.

혹시 브라우저가 아니라 서버에서 호출하는 방식이면 이런 문제가 발생할 수 있나요?