内容
# MCP Application 구현하기
MCP(Model Context Protocol)은 생성형 AI application이 외부 데이터를 활용하는 주요한 인터페이스로 빠르게 확산되고 있습니다. 2024년 11월에 Anthropic의 오픈소스 프로젝트로 시작되었고, 현재 Cursor뿐 아니라 OpenAI에서도 지원하고 있습니다. 여기에서는 [MCP with LangChain](https://github.com/langchain-ai/langchain-mcp-adapters)을 이용하여 LangGraph로 만든 application이 MCP를 활용하는 방법에 대해 설명합니다. 여기서 구현한 RAG는 Amazon의 완전관리형 RAG 서비스인 Knowledge base로 구현되었으므로, 문서의 텍스트 추출, 동기화, chunking과 같은 작업을 손쉽게 수행할 수 있으며, 멀티모달을 이용해 이미지/표를 분석할 수 있습니다. 여기에서는 MCP server에서 RAG에 손쉽게 접근할 수 있도록 AWS Lambda를 이용해 API를 구성하였습니다.
아래 architecture는 AWS 환경에서 MCP를 포함한 Agent를 구성하는것을 보여줍니다. Agent는 MCP server/client 구조를 활용하여 외부의 데이터 소스를 활용할 수 있습니다. MCP client는 MCP server와 JSON-RPC 프로토콜에 기반하여 stdio/SSE로 통신을 수행합니다. Stdio 사용시 MCP Server는 python, java와 같은 코드로 구성이 되고, client에서 요청이 오면 RAG나 인터넷등을 이용해 데이터를 수집하거나 전달하는 역할을 수행합니다. SSE로 할 경우에 MCP client와 server는 IP로 통신을 하게 됩니다. 여기서는 Streamlit을 이용해 application의 UI를 구성하고, 사용자는 ALB - CloudFront를 이용해 HTTPS 방식으로 브라우저를 통해 application을 이용합니다. 또한, 여기에서는 커스터마이징이 유리한 LangGraph를 이용해 MCP 기반의 application을 개발하는것을 설명합니다.
<img src="https://github.com/user-attachments/assets/a263f0fa-22f1-483b-8936-a59429ace173" width="750">
## MCP 활용
### MCP Basic
사용자는 자신의 Computer에 설치된 Claude Desktop, Cursor와 같은 AI 도구뿐 아니라 주로 Agent형태로 개발된 어플리케이션을 통해 MCP 서버에 연결할 수 있습니다. MCP server는 MCP client의 요청에 자신이 할수 있는 기능을 capability로 제공하고 client의 요청을 수행합니다. MCP server는 local computer의 파일이나 데이터베이스를 조회할 수 있을뿐 아니라 인터넷에 있는 외부 서버의 API를 이용해 필요한 정보를 조회할 수 있습니다. MCP Client는 Server와 JSON-RPC 2.0 프로토콜을 이용해 연결되는데, stdio나 SSE (Server-Sent Events)을 선택하여, Host의 요청을 MCP에 전달할 수 있고, 응답을 받아서 활용할 수 있습니다.
<img src="https://github.com/user-attachments/assets/31065469-69f0-4ce8-a241-747fc3504e6f" width="750">
MCP의 주요 요소의 정의와 동작은 아래와 같습니다.
- MCP Hosts: MCP 프로토콜을 통해 데이터에 접근하는 프로그램/AI 도구로서 Claude Desktop, Cursor, User Agent Application이 해당됩니다.
- MCP Clients: MCP Server와 1:1로 연결을 수행하는 Client로서 MCP Server와 stdio 또는 SSE 방식으로 연결할 수 있습니다.
- MCP Servers: 표준화된 MCP를 통해 Client에 Tool의 Capability를 알려주는 경량 프로그램으로 Local Computer의 파일이나 데이터베이스를 조회할 수 있고, 외부 API를 이용해 정보를 조회할 수 있습니다.
- Local data sources: MCP 서버가 접근할 수 있는 데이터베이스와 로컬 데이터
- Remote services: API를 통해 접근 가능한 외부 시스템
MCP를 사용하면 아래와 같은 장점이 있습니다.
- 표준화된 방식으로 다양한 데이터 소스에 접근 가능합니다.
- 애플리케이션 코드 변경 없이 MCP 서버 업데이트를 통한 새로운 기능 추가할 수 있습니다.
- 조직 전반에 걸쳐 AI 지원 및 확장이 용이합니다.
[MCP Server Components](https://www.philschmid.de/mcp-introduction)에는 아래와 같은 항목이 있습니다.
- Tools (Model-controlled): LLM이 특정 작업을 수행하기 위해 호출할 수 있는 기능(도구)으로서, API와 같이 특정한 action을 수행합니다.
```python
tools = await session.list_tools()
```
- Resources (Application-controlled): 생성형 AI 어플리케이션이 접근 할 수 있는 데이터 소스입니다. 복잡한 계산(significant computation)이나 부작용(side effect)없이 데이터를 가져올 수 있습니다.
```python
resources = await session.list_resources()
```
- Prompts (User-controlled): tool나 resource를 사용할때에 이용하는 사전 정의된 템플렛으로서 추론(inference)전에 선택할 수 있습니다.
```python
prompts = await session.list_prompts()
```
### Operation Architecture
Streamlit UI(`app.py`)에서 대화 형태·Agent 타입·MCP 서버를 선택하면 `chat.py`가 모드별로 라우팅합니다. Agent 모드는 LangGraph, Strands, Claude Agent SDK 세 가지 구현을 지원하며, MCP 서버 설정은 `mcp_config.py`에서 stdio / streamable HTTP transport로 로드됩니다.
```mermaid
flowchart TB
subgraph UI["Streamlit (app.py)"]
MODE["대화 형태 / Agent 타입"]
MCPUI["MCP 서버 선택"]
end
subgraph Router["chat.py"]
RAG[run_rag_with_knowledge_base]
LG[run_langgraph_agent]
ST[run_strands_agent]
CA[run_claude_agent]
end
subgraph LLM["Amazon Bedrock"]
BR[Bedrock Runtime]
KB[Knowledge Base]
end
subgraph Agents["Agent 구현"]
LGA["LangGraph\nMultiServerMCPClient + built-in tools"]
STA["Strands\nMCPClientManager + strands_tools"]
CLA["Claude SDK\nClaudeSDKClient + MCP"]
end
subgraph MCPServers["MCP Servers (mcp_config.py)"]
MCP["knowledge base · tavily · aws document · korea_weather · ..."]
end
subgraph Storage["Artifacts / S3"]
ART[artifacts/]
S3[(S3)]
end
MODE --> RAG
MODE --> LG
MODE --> ST
MODE --> CA
MCPUI --> MCPServers
RAG --> KB --> BR
LG --> LGA --> BR
ST --> STA --> BR
CA --> CLA --> BR
LGA --> MCPServers
STA --> MCPServers
CLA --> MCPServers
LGA --> ART
STA --> ART
LGA --> S3
STA --> S3
```
| 모드 | 모듈 | 설명 |
|------|------|------|
| 일상적인 대화 | `chat.general_conversation` | 대화 이력 + Bedrock Runtime `invoke_model_with_response_stream` 스트리밍 |
| RAG | `chat.run_rag_with_knowledge_base` | Bedrock Knowledge Base 검색(`retrieve`) 후 Bedrock Runtime으로 답변 생성 |
| **Agent** / **Agent (Chat)** — langgraph | `chat.run_langgraph_agent` | LangGraph StateGraph + LangChain `MultiServerMCPClient` + built-in tools |
| **Agent** / **Agent (Chat)** — strands | `chat.run_strands_agent` | Strands SDK + `strands_tools` + `MCPClientManager` + built-in tools |
| **Agent** / **Agent (Chat)** — claude | `claude_agent.run_claude_agent` | Claude Agent SDK(`ClaudeSDKClient`) + MCP (Bedrock 백엔드) |
| 이미지 분석 | `chat.summarize_image` / `chat.get_image_summarization` | ChatBedrock 멀티모달 (이미지 + 텍스트) 분석 |
| 번역하기 | `chat.translate_text` | 한↔영 번역 |
| 문법 검토하기 | `chat.check_grammer` | 한국어·영어 문법 검토 및 수정 제안 |
Agent (Chat) 모드는 Agent 모드와 동일한 Agent 구현을 사용하되, `history_mode=Enable`로 대화 이력을 유지합니다. MCP 서버는 사이드바에서 다중 선택 가능하며, `mcp_config.load_selected_config()`로 stdio 또는 streamable HTTP 설정이 병합됩니다.
### LangChain MCP Adapter
[LangChain MCP Adapter](https://github.com/langchain-ai/langchain-mcp-adapters)는 MCP를 LangGraph agent와 함께 사용할 수 있게 해주는 경량의 랩퍼(lightweight wrapper)로서 MIT 기반의 오픈소스입니다. MCP Adapter의 주된 역할은 MCP server를 위한 tool들을 정의하고, MCP client에서 tools의 정보를 조회하고 LangGraph의 tool node로 정의하여 활용할 수 있도록 도와줍니다.
#### 사전 준비
MCP와 LangChain MCP Adapter를 아래와 같이 설치합니다.
```text
pip install mcp langchain-mcp-adapters
```
#### MCP Server
RAG 검색을 위한 MCP server는 아래와 같이 정의할 수 있습니다. Server의 transport를 "stdio"로 지정하면 server를 지속 실행시키지 않더라도, client가 server의 python code를 직접 실행할 수 있어서 편리합니다.
```python
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name = "retrieve"
)
@mcp.tool()
def rag_search(keyword: str) -> str:
"search keyword"
return retrieve(keyword)
if __name__ =="__main__":
mcp.run(transport="stdio")
```
Server는 요청이 들어오면, retrieve_knowledge_base()로 RAG 검색을 수행합니다.
```python
bedrock_agent_runtime_client = boto3.client("bedrock-agent-runtime", region_name=bedrock_region)
def retrieve(query):
response = bedrock_agent_runtime_client.retrieve(
retrievalQuery={"text": query},
knowledgeBaseId=knowledge_base_id,
retrievalConfiguration={
"vectorSearchConfiguration": {"numberOfResults": number_of_results},
},
)
payload = json.load(output['Payload'])
return payload['response'], []
```
[Drug discovery](https://github.com/kyopark2014/mcp/blob/main/drug-discovery.md)와 관련하여, [arXiv](./application/mcp_server_arxiv.py), [ChEMBL](./application/mcp_server_chembl.py), [ClinicalTrials.gov](./application/mcp_server_clinicaltrial.py), [PubMed](./application/mcp_server_pubmed.py)을 지원하고 있습니다.
#### MCP Client
MCP client이 하나의 MCP server만 볼 경우에는 아래와 같이 stdio_client와 StdioServerParameters를 이용해 구현할 수 있습니다. MCP server에 대한 정보는 config.json에서 읽어오거나 streamlit에서 사용자가 입력한 정보를 사용할 수 있습니다. load_mcp_server_parameters()에서는 mcp_json을 읽어와서 [StdioServerParameters](https://github.com/langchain-ai/langchain-mcp-adapters)을 구성합니다.
```python
from mcp import ClientSession, StdioServerParameters
def load_mcp_server_parameters():
mcp_json = json.loads(mcp_config)
mcpServers = mcp_json.get("mcpServers")
command = ""
args = []
if mcpServers is not None:
for server in mcpServers:
config = mcpServers.get(server)
if "command" in config:
command = config["command"]
if "args" in config:
args = config["args"]
break
return StdioServerParameters(
command=command,
args=args
)
```
아래와 같이 MCP server에 대한 정보로 stdio_client를 구성합니다. 이때 tools에 대한 정보를 load_mcp_tools로 가져옵니다. Agent에서는 tool 정보를 bind하고 ainvoke를 이용해 요청된 동작을 수행합니다.
```python
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
server_params = load_mcp_server_parameters()
client = MultiServerMCPClient(server_params)
tools = await client.get_tools()
app = buildAgent(tools)
config = {
"recursion_limit": 50,
"configurable": {"thread_id": user_id},
"tools": tools,
"system_prompt": None
}
inputs = {
"messages": [HumanMessage(content=query)]
}
async for output in app.astream(inputs, config):
for key, value in output.items():
if isinstance(value, dict) and "messages" in value:
final_output = value
if final_output and "messages" in final_output and len(final_output["messages"]) > 0:
result = final_output["messages"][-1].content
else:
result = "답변을 찾지 못하였습니다."
```
여기에서는 아래와 같이 ReAct 방식의 LangGraph agent를 이용합니다.
```python
def call_model(state: State, config):
system = (
"당신의 이름은 서연이고, 질문에 친근한 방식으로 대답하도록 설계된 대화형 AI입니다."
"상황에 맞는 구체적인 세부 정보를 충분히 제공합니다."
"모르는 질문을 받으면 솔직히 모른다고 말합니다."
"한국어로 답변하세요."
)
try:
prompt = ChatPromptTemplate.from_messages(
[
("system", system),
MessagesPlaceholder(variable_name="messages"),
]
)
chain = prompt | model
response = chain.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: State) -> Literal["continue", "end"]:
messages = state["messages"]
last_message = messages[-1]
if isinstance(last_message, AIMessage) and last_message.tool_calls:
return "continue"
else:
return "end"
def buildAgent(tools):
tool_node = ToolNode(tools)
workflow = StateGraph(State)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "action",
"end": END,
},
)
workflow.add_edge("action", "agent")
return workflow.compile()
```
MCP client는 아래와 같이 실행합니다. 비동기적으로 실행하기 위해서 asyncio를 이용하였습니다.
```python
asyncio.run(mcp_agent(query, st))
```
여기서는 customize가 용이하도록 agent를 정의하였습니다.
## MCP Servers의 활용
[Model Context Protocol servers](https://github.com/modelcontextprotocol/servers)에서도 아래와 같은 서버들에 대한 정보를 제공하고 있습니다.
- [Perplexity Ask MCP Server](https://github.com/ppl-ai/modelcontextprotocol)
- [Riza MCP Server](https://github.com/riza-io/riza-mcp)
- [Tavily MCP Server](https://github.com/tavily-ai/tavily-mcp)
[Smithery](https://smithery.ai/)에서 MCP server를 찾아보고 필요한 서버를 찾으면 접속할 수 있는 MCP 서버 정보를 JSON 형태로 조회할 수 있습니다.
<img src="https://github.com/user-attachments/assets/62e534ee-88bd-4f9f-a4ff-129522fd834f" width="500">
[Smithery - Google Search Server](https://smithery.ai/server/@gradusnikov/google-search-mcp-server)에서 확인한 구글 검색용 MCP 서버 정보는 아래와 같습니다. 검색엔진 ID와 API Key를 필요로 합니다.
```java
{
"mcpServers": {
"google-search-mcp-server": {
"command": "npx",
"args": [
"-y",
"@smithery/cli@latest",
"run",
"@gradusnikov/google-search-mcp-server",
"--config",
"{\"googleCseId\":\"b5cd8c527fbd64b72\",\"googleApiKey\":\"AIzbSyDQlYpck8-9TbBSuxoew1luOGVB6unRPNk\"}"
]
}
}
}
```
아래와 같이 json 형식의 서버정보를 업데이트 할 수 있습니다. 아래에서는 [mcp-server.py](./application/mcp-server.py)에서 정의한 search를 이용하고 있습니다.
```java
{
"mcpServers": {
"search": {
"command": "python",
"args": [
"application/mcp-server.py"
]
}
}
}
```
### AWS Cost Analysis
MCP tool로서 아래와 같이 AWS cost 정보를 가져와서 분석할 수 있습니다.
```python
from datetime import datetime, timedelta
import pandas as pd
def get_cost_analysis(days: str=30):
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
ce = boto3.client('ce')
service_response = ce.get_cost_and_usage(
TimePeriod={
'Start': start_date.strftime('%Y-%m-%d'),
'End': end_date.strftime('%Y-%m-%d')
},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
)
service_costs = pd.DataFrame([
{
'SERVICE': group['Keys'][0],
'cost': float(group['Metrics']['UnblendedCost']['Amount'])
}
for group in service_response['ResultsByTime'][0]['Groups']
])
# region cost
region_response = ce.get_cost_and_usage(
TimePeriod={
'Start': start_date.strftime('%Y-%m-%d'),
'End': end_date.strftime('%Y-%m-%d')
},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[{'Type': 'DIMENSION', 'Key': 'REGION'}]
)
region_costs = pd.DataFrame([
{
'REGION': group['Keys'][0],
'cost': float(group['Metrics']['UnblendedCost']['Amount'])
}
for group in region_response['ResultsByTime'][0]['Groups']
])
# Daily Cost
daily_response = ce.get_cost_and_usage(
TimePeriod={
'Start': start_date.strftime('%Y-%m-%d'),
'End': end_date.strftime('%Y-%m-%d')
},
Granularity='DAILY',
Metrics=['UnblendedCost'],
GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
)
daily_costs = []
for time_period in daily_response['ResultsByTime']:
date = time_period['TimePeriod']['Start']
for group in time_period['Groups']:
daily_costs.append({
'date': date,
'SERVICE': group['Keys'][0],
'cost': float(group['Metrics']['UnblendedCost']['Amount'])
})
daily_costs_df = pd.DataFrame(daily_costs)
return {
'service_costs': service_costs,
'region_costs': region_costs,
'daily_costs': daily_costs_df
}
```
cost의 상기 3가지 결과를 그래프로 사용하기 위해서는 아래 패키지 설치가 필요합니다.
```text
pip install -U kaleido
```
### MCP Image Generation
[mcp_server_image_generation.py](./application/mcp_server_image_generation.py)과 같이 이미지 생성하는 tool을 등록합니다.
MCP config는 아래와 같이 설정합니다. [mcp_config.py](./application/mcp_config.py)을 참조합니다.
```java
{
"mcpServers": {
"imageGeneration": {
"command": "python",
"args": [
"application/mcp_server_image_generation.py"
]
}
}
}
```
이후 [mcp_nova_canvas.py](./application/mcp_nova_canvas.py)와 같이 이미지를 생성합니다.
```python
async def mcp_generate_image(ctx, prompt, negative_prompt, filename, width, height, quality, cfg_scale, seed, number_of_images):
"""Generate an image using Amazon Nova Canvas with text prompt."""
response = await generate_image_with_text(
prompt=prompt,
negative_prompt=negative_prompt,
filename=filename,
width=width,
height=height,
quality=quality,
cfg_scale=cfg_scale,
seed=seed,
number_of_images=number_of_images
)
return {
"url": [f'{path}' for path in response.paths]
}
```
MCP의 image_generation로 부터 얻은 결과는 아래와 같이 표시합니다. 상세한 내용은 [chat.py](./application/chat.py)을 참고하세요.
```python
def show_status_message(response, st):
image_url = []
for i, re in enumerate(response):
logger.info(f"message[{i}]: {re}")
if i==len(response)-1:
break
if isinstance(re, ToolMessage):
tool_result = json.loads(re.content)
logger.info(f"tool_result: {tool_result}")
if "url" in tool_result:
st.info(f"URL: {tool_result['url']}")
urls = tool_result['url']
for url in urls:
image_url.append(url)
st.image(url)
return image_url
```
### MCP AWS Diagram
[AWS Diagram MCP Server](https://awslabs.github.io/mcp/servers/aws-diagram-mcp-server/)을 이용하면 AWS Diagram을 그릴 수 있습니다. 상세한 내용은 [mcp_config.py](./application/mcp_config.py)을 참조합니다.
이때 사용하는 MCP Config는 아래와 같습니다.
```python
{
"mcpServers": {
"awslabs.aws-diagram-mcp-server": {
"command": "uvx",
"args": ["awslabs.aws-diagram-mcp-server"],
"env": {
"FASTMCP_LOG_LEVEL": "ERROR"
},
}
}
}
```
Diagram을 그리기 위해서는 [Graphviz](https://www.graphviz.org/download/)를 따라서 graphviz를 설치합니다. Mac에서는 아래 명령어를 사용합니다.
```text
brew install graphviz
```
### MCP AWS Documentation
[AWS Documentation MCP Server](https://awslabs.github.io/mcp/servers/aws-documentation-mcp-server/)을 이용하여 AWS 문서들을 조회할 수 있습니다. 이때 사용하는 MCP config는 아래와 같습니다. 상세한 내용은 [mcp_config.py](./application/mcp_config.py)을 참조합니다.
```java
{
"mcpServers": {
"awslabs.aws-documentation-mcp-server": {
"command": "uvx",
"args": ["awslabs.aws-documentation-mcp-server@latest"],
"env": {
"FASTMCP_LOG_LEVEL": "ERROR"
}
}
}
}
```
### 리전별 사용할 수 있는 모델의 확인 방법
사용할 수 있는 모델의 확인 방법은 아래와 같습니다.
```text
aws bedrock list-foundation-models --region=us-west-2 --by-provider anthropic --query "modelSummaries[*].modelId"
```
### Message Trim
LangGraph 에이전트([application/langgraph_agent.py](./application/langgraph_agent.py)의 `call_model`)는 LLM 호출 직전에 **HumanMessage 기준 최근 N턴**만 남깁니다. LangGraph state의 `messages`는 checkpointer에 그대로 두고, **모델에 넘기는 메시지만** trim합니다. `history_mode=Enable`/`Disable` 모두 동일하게 적용됩니다.
**기본값:** `MAX_CONTEXT_TURNS = 5` (일반 채팅의 `SimpleMemory(k=5)`와 동일한 “최근 5턴” 의도)
**설정 변경:**
- [application/langgraph_agent.py](./application/langgraph_agent.py)의 `MAX_CONTEXT_TURNS` 상수 수정
- 또는 [application/chat.py](./application/chat.py)의 `create_agent()`에서 config의 `max_turns` / `configurable.max_turns` 지정
- `max_turns=0`이면 trim 비활성화
상수와 trim 함수는 `langgraph_agent.py`에 정의합니다.
```python
# application/langgraph_agent.py
MAX_CONTEXT_TURNS = 5
def trim_messages_by_human_turns(messages: list, max_turns: int) -> list:
"""Keep messages from the last N HumanMessage turns (inclusive)."""
if max_turns <= 0 or not messages:
return messages
human_indices = [i for i, msg in enumerate(messages) if isinstance(msg, HumanMessage)]
if len(human_indices) <= max_turns:
return messages
return messages[human_indices[-max_turns]:]
```
`call_model`에서는 `ToolMessage` content 정규화 후 trim을 적용합니다.
```python
# application/langgraph_agent.py — call_model() 내부
max_turns = (
config.get("configurable", {}).get("max_turns")
or config.get("max_turns")
or MAX_CONTEXT_TURNS
)
trimmed = trim_messages_by_human_turns(messages, max_turns)
if len(trimmed) < len(messages):
logger.info(
f"trimmed messages from {len(messages)} to {len(trimmed)} "
f"(max_turns={max_turns})"
)
messages = trimmed
prompt = ChatPromptTemplate.from_messages(
[
("system", system),
MessagesPlaceholder(variable_name="messages"),
]
)
chain = prompt | model
async for chunk in chain.astream({"messages": messages}):
...
```
에이전트 config는 `chat.py`의 `create_agent()`에서 생성하며, `history_mode`와 관계없이 `max_turns`를 전달합니다.
```python
# application/chat.py — create_agent()
if history_mode == "Enable":
app = langgraph_agent.buildChatAgentWithHistory(tools)
config = {
"recursion_limit": 50,
"configurable": {
"thread_id": user_id,
"tools": tools,
"system_prompt": None,
},
"max_turns": langgraph_agent.MAX_CONTEXT_TURNS,
}
else:
app = langgraph_agent.buildChatAgent(tools)
config = {
"recursion_limit": 50,
"configurable": {
"thread_id": user_id,
"tools": tools,
"system_prompt": None,
},
"max_turns": langgraph_agent.MAX_CONTEXT_TURNS,
}
```
**`max_turns=5`의 의미**
- **사용자 HumanMessage 5개**와, 각 턴에 이어진 **모든 후속 메시지**를 유지
- 1턴 = `HumanMessage` 1개 + 그 뒤의 `AIMessage`, `ToolMessage`, 도구 feedback loop 전체
- 도구를 여러 번 호출해도 **같은 사용자 질문이면 1턴**으로 카운트
**예 (도구 사용 포함)**
```
Human(Q1) → AI(tool_calls) → ToolMessage → AI(A1)
Human(Q2) → AI(A2)
Human(Q3) → AI(tool_calls) → ToolMessage → AI(A3)
```
`max_turns=2`이면 **Q2부터** 유지:
```
Human(Q2) → AI(A2) → Human(Q3) → AI(tool_calls) → ToolMessage → AI(A3)
```
**메시지 개수 trim과의 차이**
| 방식 | `N=5`일 때 |
|------|------------|
| 이전 (메시지 개수) | 메시지 객체 5개만 유지 → 도구 루프 때문에 사용자 턴 수가 불규칙 |
| 현재 (HumanMessage 턴) | 사용자 질문 5개 + 각 턴의 AI/Tool 응답 전체 유지 |
**Checkpointer와의 관계**
- `history_mode=Enable`일 때 `MemorySaver` checkpointer에는 **전체 대화 이력**이 저장됩니다.
- trim은 LLM 컨텍스트 윈도우 관리용이며, 저장된 history를 삭제하지 않습니다.
- CloudWatch(`/ecs/...`) 또는 애플리케이션 로그에서 `trimmed messages from X to Y (max_turns=5)`로 trim 여부를 확인할 수 있습니다.
## 배포하기
### EC2로 배포하기
AWS console의 EC2로 접속하여 [Launch an instance](https://us-west-2.console.aws.amazon.com/ec2/home?region=us-west-2#Instances:)를 선택합니다. [Launch instance]를 선택한 후에 적당한 Name을 입력합니다. (예: es) key pair은 "Proceed without key pair"을 선택하고 넘어갑니다.
<img width="700" alt="ec2이름입력" src="https://github.com/user-attachments/assets/c551f4f3-186d-4256-8a7e-55b1a0a71a01" />
Instance가 준비되면 [Connet] - [EC2 Instance Connect]를 선택하여 아래처럼 접속합니다.
<img width="700" alt="image" src="https://github.com/user-attachments/assets/e8a72859-4ac7-46af-b7ae-8546ea19e7a6" />
이후 아래와 같이 python, pip, git, boto3를 설치합니다.
```text
sudo yum install python3 python3-pip git docker -y
pip install boto3
```
Workshop의 경우에 아래 형태로 된 Credential을 복사하여 EC2 터미널에 입력합니다.
<img width="700" alt="credential" src="https://github.com/user-attachments/assets/261a24c4-8a02-46cb-892a-02fb4eec4551" />
아래와 같이 git source를 가져옵니다.
```python
git clone https://github.com/kyopark2014/es-us-project
```
아래와 같이 installer.py를 이용해 설치를 시작합니다.
```python
cd es-us-project && python3 installer.py
```
API 구현에 필요한 credential은 secret으로 관리합니다. 따라서 설치시 필요한 credential 입력이 필요한데 아래와 같은 방식을 활용하여 미리 credential을 준비합니다.
- 일반 인터넷 검색: [Tavily Search](https://app.tavily.com/sign-in)에 접속하여 가입 후 API Key를 발급합니다. 이것은 tvly-로 시작합니다.
- 날씨 검색: [openweathermap](https://home.openweathermap.org/api_keys)에 접속하여 API Key를 발급합니다. 이때 price plan은 "Free"를 선택합니다.
설치가 완료되면 아래와 같은 CloudFront로 접속하여 동작을 확인합니다.
<img width="500" alt="cloudfront_address" src="https://github.com/user-attachments/assets/7ab1a699-eefb-4b55-b214-23cbeeeb7249" />
인프라가 더이상 필요없을 때에는 uninstaller.py를 이용해 제거합니다.
```text
python uninstaller.py
```
### 배포된 Application 업데이트 하기
AWS console의 EC2로 접속하여 [Launch an instance](https://us-west-2.console.aws.amazon.com/ec2/home?region=us-west-2#Instances:)를 선택하여 아래와 같이 아래와 같이 "app-for-es-us"라는 이름을 가지는 instance id를 선택합니다.
<img width="750" alt="image" src="https://github.com/user-attachments/assets/7d6d756a-03ba-4422-9413-9e4b6d3bc1da" />
[connect]를 선택한 후에 Session Manager를 선택하여 접속합니다.
<img width="700" alt="image" src="https://github.com/user-attachments/assets/d1119cd6-08fb-4d3e-b1c2-77f2d7c1216a" />
이후 아래와 같이 업데이트한 후에 다시 브라우저에서 확인합니다.
```text
cd ~/es-us-project/ && sudo ./update.sh
```
### 실행 로그 확인
[EC2 console](https://us-west-2.console.aws.amazon.com/ec2/home?region=us-west-2#Instances:)에서 "app-for-es-us"라는 이름을 가지는 instance id를 선택 한 후에, EC2의 Session Manager를 이용해 접속합니다.
먼저 아래와 같이 현재 docker container ID를 확인합니다.
```text
sudo docker ps
```
이후 아래와 같이 container ID를 이용해 로그를 확인합니다.
```text
sudo docker logs [container ID]
```
실제 실행시 결과는 아래와 같습니다.
<img width="600" src="https://github.com/user-attachments/assets/2ca72116-0077-48a0-94be-3ab15334e4dd" />
### Local에서 실행하기
AWS 환경을 잘 활용하기 위해서는 [AWS CLI를 설치](https://docs.aws.amazon.com/ko_kr/cli/v1/userguide/cli-chap-install.html)하여야 합니다. EC2에서 배포하는 경우에는 별도로 설치가 필요하지 않습니다. Local에 설치시는 아래 명령어를 참조합니다.
```text
curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install
```
AWS credential을 아래와 같이 AWS CLI를 이용해 등록합니다.
```text
aws configure
```
설치하다가 발생하는 각종 문제는 [Kiro-cli](https://aws.amazon.com/ko/blogs/korea/kiro-general-availability/)를 이용해 빠르게 수정합니다. 아래와 같이 설치할 수 있지만, Windows에서는 [Kiro 설치](https://kiro.dev/downloads/)에서 다운로드 설치합니다. 실행시는 셀에서 "kiro-cli"라고 입력합니다.
```python
curl -fsSL https://cli.kiro.dev/install | bash
```
venv로 환경을 구성하면 편리하게 패키지를 관리합니다. 아래와 같이 환경을 설정합니다.
```text
python -m venv .venv
source .venv/bin/activate
```
이후 다운로드 받은 github 폴더로 이동한 후에 아래와 같이 필요한 패키지를 추가로 설치 합니다.
```text
pip install -r requirements.txt
```
이후 아래와 같은 명령어로 streamlit을 실행합니다.
```text
streamlit run application/app.py
```
### MCP Inspector
Development Mode에서 mcp server를 테스트 하기 위해 MCP inspector를 이용할 수 있습니다. 아래와 같이 cli를 설치합니다.
```text
pip install 'mcp[cli]'
```
이후 아래와 같이 실행하면 쉽게 mcp-server.py의 동작을 테스트 할 수 있습니다. 실행시 http://localhost:5173 와 같은 URL을 제공합니다.
```text
mcp dev mcp-server.py
```
## 실행 결과
### MCP로 RAG를 조회하여 활용하기
[error_code.pdf](./contents/error_code.pdf)을 다운로드 한 후에 파일을 업로드합니다. 이후 아래와 같이 "보일러 에러중 수압과 관련된 에러 코드를 검색해주세요."와 같이 입력하면 mcp를 이용해 tool의 정보를 가져오고, search tool로 얻어진 정보를 이용해 아래와 같은 정보를 보여줄 수 있습니다. 이때 search tool은 lambda를 실행하는데 lambda에서는 완전 관리형 RAG 서비스인 knowledge base를 이용하여 검색어를 조회하고 관련성을 평가한 후에 관련된 문서만을 전달합니다. Agent는 RAG를 조회하여 얻어진 정보로 답변을 아래와 같이 구합니다.
<img src="https://github.com/user-attachments/assets/01b5e47f-ada1-405e-8455-3d3ce260cb41" width="650">
### MCP로 인터넷 검색을 하여 활용하기
[smithery-Tavily](https://smithery.ai/server/mcp-tavily)에 접속하여 환경에 맞는 설정값을 얻어옵니다. 아래는 Mac/Linux의 JSON format의 접속 정보입니다.
```java
{
"mcpServers": {
"mcp-tavily": {
"command": "npx",
"args": [
"-y",
"@smithery/cli@latest",
"run",
"mcp-tavily",
"--key",
"132c5abd-6f2e-4e42-89a1-d0b1fcb75613"
]
}
}
}
```
아래는 기본 설정된 RAG를 위한 정보입니다.
```java
{
"mcpServers": {
"search": {
"command": "python",
"args": [
"application/mcp-server.py"
]
}
}
}
```
아래는 multiple mcp server를 설정시 config 입니다.
```python
{
"mcpServers":{
"RAG":{
"command":"python",
"args":[
"application/mcp-server.py"
]
},
"mcp-tavily":{
"command":"npx",
"args":[
"-y",
"@smithery/cli@latest",
"run",
"mcp-tavily",
"--key",
"132c5abd-6f2e-4e42-89a1-d0b1fcb75613"
]
}
}
}
```
이 정보를 아래와 같이 왼쪽 메뉴의 MCP Config에 복사합니다.
<img src="https://github.com/user-attachments/assets/0e054f8e-4356-42e0-a70e-636af8d377c8" width="300">
이후 메뉴에서 "Agent"를 선택후에 아래와 같이 "강남역 맛집은?"라고 입력후 결과를 확인합니다.
<img src="https://github.com/user-attachments/assets/8275bf94-2f46-475d-9eea-02568e70199b" width="650">
### AWS Cost Analysis
"지난 한달간의 AWS 비용을 요약해주세요." 입력후에 결과를 확인합니다.
<img src="https://github.com/user-attachments/assets/5794def3-47fd-498f-8074-0d970bee1c99" width="650">
### 이미지 생성
왼쪽 메뉴에서 Agent를 선택하고, MCP config로 "image generation"을 선택한 후에 "노을이 지는 아름다운 강변을 달리기하는 사람을 그려주세요."라고 입력 후에 결과를 확인합니다.
<img src="https://github.com/user-attachments/assets/a4147b0f-86d2-4042-83bb-9c9b61e51a2f" width="650">
### AWS Architecture 그리기
메뉴에서 "Agent(Chat)"를 선택하고 MCP로 "aws diagram"를 고른 후, "Amazon S3로 web hosting을 하기 위한 architecture를 추천해주세요."와 "Cognito를 이용해 인증할수 있도록 해주세요."으로 순차적으로 명령을 하면 아래와 같은 결과를 얻을 수 있습니다.
### Code Interpreter의 활용
"strawberry의 r의 갯수는?"로 질문하면 아래와 같이 Tools 리스트에서 "repl_coder"가 선택되어 활용됩니다.
<img src="https://github.com/user-attachments/assets/2575d4b1-8871-4677-a982-82d60ec6036a" width="650">
### Storage
Tool에서 "aws storage"를 선택하고, "내 s3 전체 사용량은?"이라고 질문합니다. 이때의 결과는 아래와 같습니다.
<img src="https://github.com/user-attachments/assets/a9e41d68-6a2a-4cda-a264-58cc3124f51d" width="650">
"내 aws strorage 사용량은?"이라고 질문하면, S3, EFS, EBS의 용량을 확인하여 아래와 같이 답변할 수 있습니다.
<img src="https://github.com/user-attachments/assets/4c390cee-fd91-487e-ba0d-8b99fd7389ae" width="650">
## Reference
[MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk)
[Using MCP with LangGraph agents](https://www.youtube.com/watch?v=OX89LkTvNKQ)
[MCP From Scratch](https://mirror-feeling-d80.notion.site/MCP-From-Scratch-1b9808527b178040b5baf83a991ed3b2)
[Understanding MCP From Scratch](https://www.youtube.com/watch?v=CDjjaTALI68)
[LangChain MCP Adapters](https://github.com/langchain-ai/langchain-mcp-adapters)
["Vibe Coding" LangGraph apps with llms.txt and MCP](https://www.youtube.com/watch?v=fk2WEVZfheI)
[MCP LLMS-TXT Documentation Server](https://github.com/langchain-ai/mcpdoc)
[MCP - For Server Developers](https://modelcontextprotocol.io/quickstart/server)
[Model Context Protocol (MCP) and Amazon Bedrock](https://community.aws/content/2uFvyCPQt7KcMxD9ldsJyjZM1Wp/model-context-protocol-mcp-and-amazon-bedrock)
[Model Context Protocol servers](https://github.com/modelcontextprotocol/servers)
[Langchain.js MCP Adapter](https://www.linkedin.com/posts/langchain_mcp-adapters-released-introducing-our-activity-7308925375160467457-_BPL/?utm_source=share&utm_medium=member_android&rcm=ACoAAA5jTp0BX-JuOkof3Ak56U3VlXjQVT43NzQ)
[Using LangChain With Model Context Protocol (MCP)](https://cobusgreyling.medium.com/using-langchain-with-model-context-protocol-mcp-e89b87ee3c4c)
[Desktop Commander MCP](https://github.com/wonderwhy-er/DesktopCommanderMCP)
[Smithery](https://smithery.ai/)
[Cursor AI 말고, 나만의 #MCP 에이전트 앱 만들어 보기!](https://www.youtube.com/watch?v=ISrYHGg2C2c)
[The Top 7 MCP-Supported AI Frameworks](https://medium.com/@amosgyamfi/the-top-7-mcp-supported-ai-frameworks-a8e5030c87ab)
[Drug Discovery Agent based on Amazon Bedrock](https://github.com/hsr87/drug-discovery-agent)
[Price Per Token](https://pricepertoken.com/)
[Claude Code Subagents Collection](https://github.com/wshobson/agents)
[Top 5 Open-Source Agentic Frameworks](https://research.aimultiple.com/agentic-frameworks/)
MCP 市场
