MCP RAG Server
About
A lightweight Python server for Retrieval-Augmented Generation (RAG) using AWS Lambda. It retrieves knowledge from external data sources like arXiv and PubMed.
Details
- Author
- kyopark2014
- Categories
- Developer Tools, Knowledge Base, Other, AI
Jump to
Setup
Install MCP RAG Server in your MCP client (Claude Desktop, Cursor, Windsurf, and others).
Repository: https://github.com/kyopark2014/mcp
Follow the installation instructions in the repository README, then restart your MCP client.
MCP(Model Context Protocol)은 생성형 AI application이 외부 데이터를 활용하는 주요한 인터페이스로 빠르게 확산되고 있습니다. 2024년 11월에 Anthropic의 오픈소스 프로젝트로 시작되었고, 현재 Cursor뿐 아니라 OpenAI에서도 지원하고 있습니다. 여기에서는MCP with LangChain을 이용하여 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을 개발하는것을 설명합니다.
사용자는 자신의 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에 전달할 수 있고, 응답을 받아서 활용할 수 있습니다.
- 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 서버 업데이트를 통한 새로운 기능 추가할 수 있습니다.
- 조직 전반에 걸쳐 AI 지원 및 확장이 용이합니다.
MCP Server Components에는 아래와 같은 항목이 있습니다.
- Tools (Model-controlled): LLM이 특정 작업을 수행하기 위해 호출할 수 있는 기능(도구)으로서, API와 같이 특정한 action을 수행합니다.
- Resources (Application-controlled): 생성형 AI 어플리케이션이 접근 할 수 있는 데이터 소스입니다. 복잡한 계산(significant computation)이나 부작용(side effect)없이 데이터를 가져올 수 있습니다.
resources = await session.list_resources()
- Prompts (User-controlled): tool나 resource를 사용할때에 이용하는 사전 정의된 템플렛으로서 추론(inference)전에 선택할 수 있습니다.
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
Agent (Chat) 모드는 Agent 모드와 동일한 Agent 구현을 사용하되,history_mode=Enable로 대화 이력을 유지합니다. MCP 서버는 사이드바에서 다중 선택 가능하며,mcp_config.load_selected_config()로 stdio 또는 streamable HTTP 설정이 병합됩니다.
MCP와 LangChain MCP Adapter를 아래와 같이 설치합니다.
RAG 검색을 위한 MCP server는 아래와 같이 정의할 수 있습니다. Server의 transport를 "stdio"로 지정하면 server를 지속 실행시키지 않더라도, client가 server의 python code를 직접 실행할 수 있어서 편리합니다.
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 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를 구성합니다. 이때 tools에 대한 정보를 load_mcp_tools로 가져옵니다. Agent에서는 tool 정보를 bind하고 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를 이용합니다.
…
Sign in to leave a review
Use Google, GitHub, or an email account so ratings stay tied to real people.
No reviews posted yet.
