MCP로 RAG Application 구현하기

实现 MCP 应用

MCP（Model Context Protocol）正迅速成为生成式 AI 应用利用外部数据的主要接口。它于 2024 年 11 月作为 Anthropic 的开源项目启动，目前不仅受到 Cursor 支持，也受到 OpenAI 支持。本文将介绍如何使用 MCP with LangChain，让使用 LangGraph 构建的应用利用 MCP。本文实现的 RAG 基于 Amazon 的全托管 RAG 服务 Knowledge base，因此可以轻松执行文档文本提取、同步和分块等工作，还可以利用多模态功能分析图像和表格。本文使用 AWS Lambda 构建 API，以便通过 MCP 服务轻松访问 RAG。

下面的架构展示了如何在 AWS 环境中构建包含 MCP 的 Agent。Agent 可以利用 MCP 服务端/客户端结构访问外部数据源。MCP 客户端基于 JSON-RPC 协议，通过 stdio/SSE 与 MCP 服务通信。使用 Stdio 时，MCP 服务由 python、java 等代码构建；收到客户端请求后，服务会利用 RAG 或互联网等收集或传递数据。使用 SSE 时，MCP 客户端和服务端通过 IP 通信。本文使用 Streamlit 构建应用的 UI，用户通过 ALB - CloudFront，以 HTTPS 方式在浏览器中使用应用。此外，本文还将介绍如何使用便于自定义的 LangGraph 开发基于 MCP 的应用。

使用 MCP

MCP 基础

用户不仅可以通过安装在自己 Computer 上的 Claude Desktop、Cursor 等 AI 工具连接到 MCP 服务端，也可以通过主要以 Agent 形式开发的应用连接。MCP 服务端会将自身能够执行的功能作为 capability 提供给 MCP 客户端，并执行客户端的请求。MCP 服务端不仅可以查询本地计算机上的文件或数据库，也可以利用互联网外部服务的 API 查询所需信息。MCP 客户端使用 JSON-RPC 2.0 协议连接到服务端，可以选择 stdio 或 SSE（Server-Sent Events），将 Host 的请求传递给 MCP，并接收和使用响应。

MCP 的主要组成部分及其工作方式定义如下。

• MCP Hosts：通过 MCP 协议访问数据的程序/AI 工具，包括 Claude Desktop、Cursor 和 User Agent Application。
• MCP Clients：与 MCP 服务端进行一对一连接的客户端，可以通过 stdio 或 SSE 方式连接到 MCP 服务端。
• MCP Servers：通过标准化的 MCP 向客户端提供工具 capability 的轻量级程序，可以查询本地计算机上的文件或数据库，也可以利用外部 API 查询信息。
• Local data sources：MCP 服务端可以访问的数据库和本地数据
• Remote services：可通过 API 访问的外部系统

使用 MCP 有以下优点。

• 可以通过标准化的方式访问各种数据源。
• 无需更改应用代码，即可通过更新 MCP 服务端添加新功能。
• 便于在整个组织范围内提供和扩展 AI 支持。

MCP Server Components包括以下项目。

• Tools (Model-controlled)：LLM 为执行特定任务而调用的功能（工具），可像 API 一样执行特定 action。

tools = await session.list_tools()

• Resources (Application-controlled)：生成式 AI 应用可以访问的数据源。无需进行复杂计算（significant computation）或产生副作用（side effect）即可获取数据。

resources = await session.list_resources()

• Prompts (User-controlled)：使用 tool 或 resource 时使用的预定义模板，可在推理（inference）之前选择。

prompts = await session.list_prompts()

运行架构

在 Streamlit UI（app.py）中选择对话形式、Agent 类型和 MCP 服务后，chat.py 会按模式进行路由。Agent 模式支持 LangGraph、Strands 和 Claude Agent SDK 三种实现，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 + 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 是一个轻量级封装（lightweight wrapper），可让 MCP 与 LangGraph agent 配合使用，基于 MIT 许可证开源。MCP Adapter 的主要作用是为 MCP server 定义工具，并帮助 MCP client 查询工具信息，将其定义为 LangGraph 的 tool node 以供使用。

准备工作

按以下方式安装 MCP 和 LangChain MCP Adapter。

pip install mcp langchain-mcp-adapters

MCP Server

可以按以下方式定义用于 RAG 搜索的 MCP server。将 server 的 transport 指定为 "stdio" 后，即使不持续运行 server，client 也可以直接执行 server 的 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 搜索。

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 client 只需访问一个 MCP server，可以按以下方式使用 stdio_client 和 StdioServerParameters 实现。MCP server 的信息可以从 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 server 的信息配置 stdio_client。此时，通过 load_mcp_tools 获取 tools 的信息。在 Agent 中绑定 tool 信息，并使用 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 agent。

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。

asyncio.run(mcp_agent(query, st))

这里定义了 agent，以便于自定义。

MCP 服务的应用

Model Context Protocol servers 也提供了以下服务器的信息。

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

可以在 Smithery 中查找 MCP server。找到所需服务器后，可以查询用于连接的 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 成本分析

可以按以下方式使用 MCP tool 获取并分析 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 项 cost 结果绘制成图表，需要安装以下软件包。

pip install -U kaleido

MCP 图像生成

注册用于生成图像的工具，例如 mcp_server_image_generation.py。

MCP 配置如下所示。请参阅 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 Diagram

使用 AWS Diagram MCP Server 可以绘制 AWS Diagram。详细内容请参阅 mcp_config.py。

此时使用的 MCP Config 如下所示。

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

要绘制 Diagram，请按照 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 Agent（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}):
            ...

Agent 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 的含义

• 保留 5 条用户 HumanMessage，以及每一轮后续的所有消息
• 1 轮 = 1 条 HumanMessage，加上其后的 AIMessage、ToolMessage 和整个工具反馈循环
• 即使多次调用工具，同一个用户问题也只计为 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]

实际运行结果如下。

在本地运行

要充分利用 AWS 环境，需要安装 AWS CLI。如果在 EC2 上部署，则无需单独安装。在本地安装时，请参考以下命令。

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 安装下载并安装。运行时，在 shell 中输入 "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 inspector 测试 MCP 服务。按如下方式安装 cli。

pip install 'mcp[cli]'

之后按如下方式运行，即可轻松测试 mcp-server.py 的运行情况。运行时会提供类似 http://localhost:5173 的 URL。

mcp dev mcp-server.py

运行结果

使用 MCP 查询和利用 RAG

下载 error_code.pdf 后上传文件。之后输入“请搜索锅炉错误中与水压相关的错误代码。”，即可通过 MCP 获取工具信息，并利用 search 工具获取的信息显示如下内容。此时，search 工具会运行 lambda；lambda 使用完全托管的 RAG 服务 knowledge base 查询搜索词、评估相关性，然后仅传递相关文档。Agent 会根据查询 RAG 获得的信息生成如下回答。

### 使用 MCP 搜索互联网并加以利用

访问 smithery-Tavily，获取适合您环境的配置值。以下是 Mac/Linux 的 JSON format 连接信息。

{
  "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"
      ]
    }
  }
}

以下是配置 multiple mcp server 时的 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 成本分析

输入“请总结过去一个月的 AWS 费用。”并查看结果。

图像生成

在左侧菜单中选择 Agent，在 MCP config 中选择“image generation”，然后输入“请画一个在夕阳西下的美丽河畔跑步的人。”并查看结果。

绘制 AWS 架构图

在菜单中选择“Agent(Chat)”，在 MCP 中选择“aws diagram”，然后依次输入“请推荐一个使用 Amazon S3 托管网站的架构。”和“请使用 Cognito 实现身份验证。”，即可获得如下结果。

使用 Code Interpreter

输入“strawberry 中有几个 r？”后，Tools 列表中的“repl_coder”会被选中并调用，如下所示。

存储

在 Tool 中选择“aws storage”，然后提问“我的 S3 总用量是多少？”。结果如下所示。

提问“我的 AWS 存储用量是多少？”后，可以检查 S3、EFS 和 EBS 的容量，并像下面这样回答。

参考资料

MCP Python SDK

在 LangGraph Agent 中使用 MCP

从零开始了解 MCP

从零开始理解 MCP

LangChain MCP 适配器

使用 llms.txt 和 MCP 对 LangGraph 应用进行“氛围编程”

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 Agent 应用！

支持 MCP 的 7 大 AI 框架

基于 Amazon Bedrock 的药物发现 Agent

每 Token 价格

Claude Code 子 Agent 集合

5 大开源 Agent 框架