MCP로 RAG Application 구현하기

MCP アプリケーションの実装

MCP（Model Context Protocol）は、生成 AI アプリケーションが外部データを活用するための主要なインターフェースとして、急速に普及しています。2024年11月に Anthropic のオープンソースプロジェクトとして始まり、現在は Cursor だけでなく OpenAI でもサポートされています。ここでは、MCP with LangChain を利用して、LangGraph で作成したアプリケーションが MCP を活用する方法について説明します。ここで実装する RAG は、Amazon の完全マネージド型 RAG サービスである Knowledge base で実装されているため、ドキュメントのテキスト抽出、同期、チャンク化などの作業を簡単に行えるほか、マルチモーダル機能を利用して画像や表を分析できます。ここでは、MCP サーバーから RAG に簡単にアクセスできるよう、AWS Lambda を利用して API を構成しました。

以下のアーキテクチャは、AWS 環境で MCP を含むエージェントを構成する方法を示しています。エージェントは MCP サーバー／クライアント構造を利用して、外部のデータソースを活用できます。MCP クライアントは MCP サーバーと JSON-RPC プロトコルに基づき、stdio／SSE で通信します。stdio を使用する場合、MCP サーバーは Python や Java などのコードで構成され、クライアントからリクエストを受けると、RAG やインターネットなどを利用してデータを収集または提供します。SSE を使用する場合、MCP クライアントとサーバーは IP で通信します。ここでは Streamlit を利用してアプリケーションの UI を構成し、ユーザーは ALB - CloudFront を利用して、HTTPS でブラウザーからアプリケーションを使用します。また、ここではカスタマイズに適した LangGraph を利用した、MCP ベースのアプリケーションの開発方法について説明します。

MCP の活用

MCP の基本

ユーザーは、自分のコンピューターにインストールした Claude Desktop や Cursor などの AI ツールだけでなく、主にエージェント形式で開発されたアプリケーションを通じて、MCP サーバーに接続できます。MCP サーバーは、MCP クライアントからのリクエストに対して、自身が実行できる機能を capability として提供し、クライアントのリクエストを実行します。MCP サーバーは、ローカルコンピューターのファイルやデータベースを参照できるだけでなく、インターネット上にある外部サーバーの API を利用して必要な情報を取得できます。MCP クライアントは、JSON-RPC 2.0 プロトコルを利用してサーバーに接続します。stdio または SSE（Server-Sent Events）を選択して、ホストのリクエストを MCP に渡し、応答を受け取って活用できます。

MCP の主な要素の定義と動作は以下のとおりです。

• MCP Hosts：MCP プロトコルを通じてデータにアクセスするプログラム／AI ツールで、Claude Desktop、Cursor、ユーザーエージェントアプリケーションが該当します。
• MCP Clients：MCP サーバーと1対1で接続するクライアントで、MCP サーバーに stdio または SSE 方式で接続できます。
• MCP Servers：標準化された MCP を通じてクライアントにツールの Capability を通知する軽量プログラムです。ローカルコンピューターのファイルやデータベースを参照できるほか、外部 API を利用して情報を取得できます。
• Local data sources：MCP サーバーがアクセスできるデータベースとローカルデータ
• Remote services：API を通じてアクセスできる外部システム

MCP を使用すると、以下のようなメリットがあります。

• 標準化された方法で、さまざまなデータソースにアクセスできます。
• アプリケーションコードを変更せずに、MCP サーバーを更新して新しい機能を追加できます。
• 組織全体で AI の導入と拡張を容易に行えます。

MCP Server Components には、以下のような項目があります。

• Tools（Model-controlled）：LLM が特定のタスクを実行するために呼び出せる機能（ツール）で、API のように特定のアクションを実行します。

tools = await session.list_tools()

• Resources（Application-controlled）：生成 AI アプリケーションがアクセスできるデータソースです。複雑な計算（significant computation）や副作用（side effect）を伴わずにデータを取得できます。

resources = await session.list_resources()

• Prompts（User-controlled）：ツールやリソースを使用する際に使う、事前に定義されたテンプレートです。推論（inference）の前に選択できます。

prompts = await session.list_prompts()

Operation Architecture

Streamlit UI（app.py）で会話形式・エージェントの種類・MCP サーバーを選択すると、chat.pyがモードごとにルーティングします。エージェントモードでは、LangGraph、Strands、Claude Agent SDKの3種類の実装をサポートしており、MCP サーバーの設定はmcp_config.pyからstdio／streamable HTTP transportで読み込まれます。

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＋組み込みツール
Agent / Agent (Chat) — strands	chat.run_strands_agent	Strands SDK＋strands_tools＋MCPClientManager＋組み込みツール
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モードと同じエージェント実装を使用しますが、history_mode=Enableにより会話履歴を保持します。サイドバーではMCP サーバーを複数選択でき、mcp_config.load_selected_config()によってstdioまたはstreamable HTTPの設定が統合されます。

LangChain MCP Adapter

LangChain MCP Adapterは、MCPをLangGraphエージェントとともに使用できるようにする軽量なラッパー（lightweight wrapper）であり、MITライセンスのオープンソースです。MCP Adapterの主な役割は、MCP サーバー用のツールを定義し、MCP クライアントからツールの情報を取得してLangGraphのツールノードとして定義し、利用できるようにすることです。

事前準備

以下のようにMCPとLangChain MCP Adapterをインストールします。

pip install mcp langchain-mcp-adapters

MCP Server

RAG検索用のMCP サーバーは、以下のように定義できます。サーバーのtransportを「stdio」に指定すると、サーバーを常時実行しておかなくても、クライアントがサーバーの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")

サーバーはリクエストを受け取ると、retrieve_knowledge_base()でRAG検索を実行します。

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関連では、arXiv、ChEMBL、ClinicalTrials.gov、PubMedをサポートしています。

MCP Client

MCP クライアントが単一のMCP サーバーのみを参照する場合は、以下のようにstdio_clientとStdioServerParametersを使用して実装できます。MCP サーバーの情報はconfig.jsonから読み込むか、Streamlitでユーザーが入力した情報を使用できます。load_mcp_server_parameters()では、mcp_jsonを読み込んでStdioServerParametersを構成します。

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 サーバーの情報を使ってstdio_clientを構成します。このとき、load_mcp_toolsでツールの情報を取得します。エージェントではツール情報をバインドし、ainvokeを使ってリクエストされた処理を実行します。

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エージェントを使用します。

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 クライアントは以下のように実行します。非同期で実行するためにasyncioを使用しています。

asyncio.run(mcp_agent(query, st))

ここでは、カスタマイズしやすいようにエージェントを定義しています。

MCP Serversの活用

Model Context Protocol serversでも、以下のようなサーバーの情報が提供されています。

• Perplexity Ask MCP Server
• Riza MCP Server
• Tavily MCP Server

SmitheryでMCP サーバーを探し、必要なサーバーが見つかったら、接続に使用できるMCP サーバーの情報をJSON形式で確認できます。

Smithery - Google Search Serverで確認した、Google検索用MCP サーバーの情報は以下のとおりです。検索エンジンIDとAPI Keyが必要です。

{
  "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で定義したsearchを使用しています。

{
  "mcpServers": {
    "search": {
      "command": "python",
      "args": [
        "application/mcp-server.py"
      ]
    }
  }
}

AWS Cost Analysis

MCP ツールとして、以下のようにAWSのコスト情報を取得して分析できます。

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
    }

コストに関する上記3つの結果をグラフとして使用するには、以下のパッケージをインストールする必要があります。

pip install -U kaleido

MCP 이미지 생성

mcp_server_image_generation.py와 같이 이미지를 생성하는ツール을登録합니다。

MCP config는 다음과 같이 설정합니다. mcp_config.py을 참조하세요.

{
    "mcpServers": {
        "imageGeneration": {
            "command": "python",
            "args": [
                "application/mcp_server_image_generation.py"
            ]
        }
    }
}

이후 mcp_nova_canvas.py와 같이 이미지를 생성합니다.

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을 참조하세요.

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 다이어그램

AWS Diagram MCP Server를 이용하면 AWS 다이어그램을 그릴 수 있습니다. 자세한 내용은 mcp_config.py을 참조하세요.

이때 사용하는 MCP Config는 다음과 같습니다.

{
    "mcpServers": {
        "awslabs.aws-diagram-mcp-server": {
            "command": "uvx",
            "args": ["awslabs.aws-diagram-mcp-server"],
            "env": {
                "FASTMCP_LOG_LEVEL": "ERROR"
            },
        }
    }
}

다이어그램을 그리려면 Graphviz를 따라 graphviz를 설치합니다. Mac에서는 다음 명령어를 사용합니다.

brew install graphviz

MCP AWS 문서

AWS Documentation MCP Server를 이용해 AWS 문서를 조회할 수 있습니다. 이때 사용하는 MCP config는 다음과 같습니다. 자세한 내용은 mcp_config.py을 참조하세요.

{
    "mcpServers": {
        "awslabs.aws-documentation-mcp-server": {
            "command": "uvx",
            "args": ["awslabs.aws-documentation-mcp-server@latest"],
            "env": {
                "FASTMCP_LOG_LEVEL": "ERROR"
            }
        }
    }
}

리전별 사용 가능한 모델 확인 방법

사용 가능한 모델을 확인하는 방법은 다음과 같습니다.

aws bedrock list-foundation-models --region=us-west-2 --by-provider anthropic --query "modelSummaries[*].modelId"

메시지 트림

LangGraph 에이전트(application/langgraph_agent.py의 call_model)는 LLM을 호출하기 직전에 HumanMessage 기준으로 최근 N턴만 남깁니다. LangGraph state의 messages는 checkpointer에 그대로 두고, 모델에 전달하는 메시지만 트림합니다. history_mode=Enable/Disable 모두 동일하게 적용됩니다.

기본값: MAX_CONTEXT_TURNS = 5 (일반 채팅의 SimpleMemory(k=5)와 같은 “최근 5턴”을 의미합니다)

설정 변경:

• application/langgraph_agent.py의 MAX_CONTEXT_TURNS 상수를 수정합니다.
• 또는 application/chat.py의 create_agent()에서 config의 max_turns / configurable.max_turns를 지정합니다.
• max_turns=0이면 트림이 비활성화됩니다.

상수와 트림 함수는 langgraph_agent.py에 정의합니다.

# 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를 정규화한 후 트림을 적용합니다.

# 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를 전달합니다.

# 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)

메시지 개수 기준 트림과의 차이

방식	N=5일 때
이전 (메시지 개수)	메시지 객체 5개만 유지 → 도구 루프 때문에 사용자 턴 수가 불규칙해짐
현재 (HumanMessage 턴)	사용자 질문 5개 + 각 턴의 AI/Tool 응답 전체 유지

Checkpointer와의 관계

• history_mode=Enable일 때 MemorySaver checkpointer에는 전체 대화 이력이 저장됩니다.
• 트림은 LLM 컨텍스트 윈도우 관리용이며, 저장된 history를 삭제하지 않습니다.
• CloudWatch(/ecs/...) 또는 애플리케이션 로그에서 trimmed messages from X to Y (max_turns=5)로 트림 여부를 확인할 수 있습니다.

배포하기

EC2にデプロイする

AWS consoleのEC2にアクセスし、Launch an instanceを選択します。[Launch instance]を選択した後、適切なNameを入力します。（例：es）key pairは「Proceed without key pair」を選択して進みます。

Instanceの準備ができたら、[Connet] - [EC2 Instance Connect]を選択して、以下のように接続します。

次に、以下のようにpython、pip、git、boto3をインストールします。

sudo yum install python3 python3-pip git docker -y
pip install boto3

Workshopの場合は、以下の形式のCredentialをコピーしてEC2ターミナルに入力します。

以下のようにgit sourceを取得します。

git clone https://github.com/kyopark2014/es-us-project

以下のようにinstaller.pyを使ってインストールを開始します。

cd es-us-project && python3 installer.py

APIの実装に必要なcredentialはsecretとして管理します。そのため、インストール時にcredentialの入力が必要です。以下の方法を利用して、あらかじめcredentialを準備します。

• 一般的なインターネット検索：Tavily Searchにアクセスして登録し、API Keyを発行します。これはtvly-で始まります。
• 天気検索：openweathermapにアクセスしてAPI Keyを発行します。このとき、price planは「Free」を選択します。

インストールが完了したら、以下のようなCloudFrontにアクセスして動作を確認します。

インフラが不要になったら、uninstaller.pyを使って削除します。

python uninstaller.py

デプロイ済みのApplicationを更新する

AWS consoleのEC2にアクセスし、Launch an instanceを選択して、以下のように「app-for-es-us」という名前のinstance idを選択します。

[connect]を選択した後、Session Managerを選択して接続します。

次に、以下のように更新した後、ブラウザーで再度確認します。

cd ~/es-us-project/ && sudo ./update.sh

実行ログを確認する

EC2 consoleで「app-for-es-us」という名前のinstance idを選択し、EC2のSession Managerを使って接続します。

まず、以下のように現在のdocker container IDを確認します。

sudo docker ps

次に、以下のようにcontainer IDを使ってログを確認します。

sudo docker logs [container ID]

実際の実行結果は以下のとおりです。

Localで実行する

AWS環境を適切に利用するには、AWS CLIをインストールする必要があります。EC2にデプロイする場合は、別途インストールする必要はありません。Localにインストールする場合は、以下のコマンドを参照してください。

curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip" 
unzip awscliv2.zip
sudo ./aws/install

以下のようにAWS CLIを使ってAWS credentialを登録します。

aws configure

インストール中に発生するさまざまな問題は、Kiro-cliを使ってすばやく修正します。以下のようにインストールできますが、WindowsではKiroのインストールからダウンロードしてインストールします。実行時はセルに「kiro-cli」と入力します。

curl -fsSL https://cli.kiro.dev/install | bash

venvで環境を構成すると、パッケージを簡単に管理できます。以下のように環境を設定します。

python -m venv .venv
source .venv/bin/activate

次に、ダウンロードしたgithubフォルダーに移動し、以下のように必要なパッケージを追加でインストールします。

pip install -r requirements.txt

次に、以下のコマンドでstreamlitを実行します。

streamlit run application/app.py

MCP Inspector

Development ModeでMCP サーバーをテストするには、MCP Inspectorを利用できます。以下のようにcliをインストールします。

pip install 'mcp[cli]'

次に、以下のように実行すると、mcp-server.pyの動作を簡単にテストできます。実行すると、http://localhost:5173のようなURLが表示されます。

mcp dev mcp-server.py

実行結果

MCPでRAGを検索して活用する

error_code.pdfをダウンロードしてから、ファイルをアップロードします。次に、以下のように「ボイラーのエラーのうち、水圧に関連するエラーコードを検索してください。」と入力すると、MCPを使ってツールの情報を取得し、search toolで得た情報を利用して、以下のような情報を表示できます。このとき、search toolはlambdaを実行します。lambdaでは、フルマネージド型RAGサービスであるknowledge baseを使って検索語を検索し、関連性を評価した後、関連する文書のみを渡します。エージェントはRAGを検索して得られた情報をもとに、以下のように回答を生成します。

### MCPでインターネット検索を活用する

smithery-Tavilyにアクセスして、お使いの環境に合った設定値を取得します。以下は、Mac/Linux用のJSON形式の接続情報です。

{
  "mcpServers": {
    "mcp-tavily": {
      "command": "npx",
      "args": [
        "-y",
        "@smithery/cli@latest",
        "run",
        "mcp-tavily",
        "--key",
        "132c5abd-6f2e-4e42-89a1-d0b1fcb75613"
      ]
    }
  }
}

以下は、デフォルトで設定されているRAG用の情報です。

{
  "mcpServers": {
    "search": {
      "command": "python",
      "args": [
        "application/mcp-server.py"
      ]
    }
  }
}

以下は、複数のMCPサーバーを設定する場合のconfigです。

{
   "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にコピーします。

次に、メニューから「Agent」を選択し、以下のように「江南駅のおいしいお店は？」と入力して、結果を確認します。

AWS Cost Analysis

「過去1か月間のAWS費用を要約してください。」と入力して、結果を確認します。

画像生成

左側のメニューからAgentを選択し、MCP configで「image generation」を選択してから、「夕日が沈む美しい川辺を走る人を描いてください。」と入力し、結果を確認します。

AWSアーキテクチャの作成

メニューから「Agent(Chat)」を選択し、MCPで「aws diagram」を選びます。次に、「Amazon S3でWebホスティングを行うためのアーキテクチャを提案してください。」「Cognitoを使って認証できるようにしてください。」と順番に指示すると、以下のような結果が得られます。

Code Interpreterの活用

「strawberryのrの数はいくつですか？」と質問すると、以下のようにToolsリストから「repl_coder」が選択され、使用されます。

Storage

Toolで「aws storage」を選択し、「自分のS3の全使用量は？」と質問します。このときの結果は以下のとおりです。

「自分のAWSストレージ使用量は？」と質問すると、S3、EFS、EBSの容量を確認して、以下のように回答できます。

参考資料

MCP Python SDK

LangGraphエージェントでMCPを使う

MCPをゼロから学ぶ

MCPをゼロから理解する

LangChain MCPアダプター

llms.txtとMCPを使ったLangGraphアプリの「Vibe Coding」

MCP LLMS-TXTドキュメントサーバー

MCP - サーバー開発者向け

Model Context Protocol（MCP）とAmazon Bedrock

Model Context Protocolサーバー

Langchain.js MCPアダプター

LangChainでModel Context Protocol（MCP）を使う

Desktop Commander MCP

Smithery

Cursor AIではなく、自分だけの#MCPエージェントアプリを作ってみよう！

MCP対応AIフレームワークのトップ7

Amazon Bedrockをベースにした創薬エージェント

トークンあたりの価格

Claude Codeサブエージェント集

オープンソースのエージェント型フレームワークトップ5