Skip to main content
Glama
deenesjakoruzh
johnylira

rag-mcp

by johnylira
OverviewSchemaSourceRelated ServersDiscussionsAdmin
Python
Hybrid

RAG-MCP Server

Model Context Protocol (MCP) 서버로, Flask, LangGraph, LlamaIndex를 사용하여 RAG를 구현합니다. 최소한의 구성으로 동작하며 확장이 용이하도록 설계되었습니다.

개요

  • HTTP 기반 MCP: POST /mcp (JSON-RPC 2.0).

  • LangGraph: 함수 호출(function calling)로 에이전트를 오케스트레이션합니다.

  • LlamaIndex: 로컬 문서를 인덱싱하고 RAG를 통해 답변합니다.

  • Docker: docker compose up --build 한 번으로 실행됩니다.

Related MCP server: DocAgent-MCP

아키텍처

MCP Client  →  Flask /mcp  →  LangGraph Agent  →  Tools
                                                ├─ rag_search (LlamaIndex RAG)
                                                ├─ agora (UTC datetime)
                                                └─ calcular (arithmetic expression)

  • 클라이언트는 도구 목록(tools/list)을 조회하고 도구를 호출(tools/call)합니다.

  • ask_agent 도구는 LangGraph 그래프를 트리거하며, LLM이 함수 호출 사용 시점을 결정합니다.

  • 다른 도구들은 MCP에서 직접 호출됩니다.

사전 요구 사항

  • Docker 및 Docker Compose (또는 로컬 Python 3.11+).

  • OPENAI_API_KEY (LLM 및 임베딩용).

프로젝트 구조

.
├── app.py              # MCP server + LangGraph + LlamaIndex
├── data/               # RAG corpus (.md, .txt, .pdf, etc.)
│   └── kb.md
├── requirements.txt    # Python dependencies
├── Dockerfile
├── docker-compose.yml
├── .env                # OPENAI_API_KEY and variables
└── README.md

설정

1. 프로젝트 클론 / 생성

빈 디렉터리를 만들고 프로젝트 파일을 붙여넣습니다 (Files 섹션 참조).

2. 환경 변수

루트에 .env 파일을 생성합니다:

OPENAI_API_KEY=sk-...
LLM_MODEL=gpt-4o-mini

지원되는 변수:

변수

기본값

설명

OPENAI_API_KEY

(필수)

OpenAI API 키.

LLM_MODEL

gpt-4o-mini

LLM 및 임베딩용 모델.

DATA_DIR

/app/data

RAG용 문서가 있는 디렉터리.

PORT

8080

서버 포트.

3. RAG 데이터

data/에 문서를 배치합니다 (예: kb.md, policies.md, manuals/). 서버는 시작 시 모든 문서를 인덱싱합니다.

최소 예시 (data/kb.md):

# Acme Corp
Support SLA: 4 hours during business hours (UTC-3).
Pro Plan costs USD 49/month and includes 10k RAG queries/day.
P1 incidents must be opened in #sre channel.

실행

Docker (권장)

docker compose up --build

서버는 http://127.0.0.1:8080에서 실행됩니다.

로컬 (Docker 없이)

pip install -r requirements.txt
export OPENAI_API_KEY=sk-...
export DATA_DIR=./data
python app.py

엔드포인트

POST /mcp (JSON-RPC 2.0)

MCP는 세 가지 주요 메서드를 사용합니다:

initialize

curl -s http://127.0.0.1:8080/mcp -H 'content-type: application/json' -d '{
  "jsonrpc":"2.0","id":1,"method":"initialize","params":{}
}'

응답:

{
  "jsonrpc":"2.0",
  "id":1,
  "result":{
    "protocolVersion":"2024-11-05",
    "capabilities":{"tools":{}},
    "serverInfo":{"name":"rag-mcp","version":"1.0.0"}
  }
}

tools/list

사용 가능한 모든 도구를 나열합니다 (ask_agent 포함):

curl -s http://127.0.0.1:8080/mcp -H 'content-type: application/json' -d '{
  "jsonrpc":"2.0","id":2,"method":"tools/list","params":{}
}'

tools/call

도구를 호출합니다:

curl -s http://127.0.0.1:8080/mcp -H 'content-type: application/json' -d '{
  "jsonrpc":"2.0","id":3,"method":"tools/call",
  "params":{
    "name":"ask_agent",
    "arguments":{"question":"What is the SLA and how much does Pro cost?"}
  }
}'

응답:

{
  "jsonrpc":"2.0",
  "id":3,
  "result":{
    "content":[{"type":"text","text":"The SLA is 4 hours during business hours (UTC-3). The Pro Plan costs USD 49/month..."}]
  }
}

GET /health

간단한 상태 확인:

curl -s http://127.0.0.1:8080/health
# {"ok": true}

사용 가능한 도구

도구

설명

InputSchema

ask_agent

RAG + 함수 호출을 사용하는 LangGraph 에이전트.

{"question": "string"}

rag_search

RAG(LlamaIndex)를 통해 베이스에서 사실을 검색합니다.

{"query": "string"}

agora

현재 UTC 날짜/시간을 반환합니다 (ISO-8601).

{}

calcular

안전한 산술 표현식을 평가합니다.

{"expressao": "string"}

직접 사용 예시

# RAG direct
curl -s http://127.0.0.1:8080/mcp -H 'content-type: application/json' -d '{
  "jsonrpc":"2.0","id":4,"method":"tools/call",
  "params":{"name":"rag_search","arguments":{"query":"support SLA"}}
}'

# Datetime
curl -s http://127.0.0.1:8080/mcp -H 'content-type: application/json' -d '{
  "jsonrpc":"2.0","id":5,"method":"tools/call",
  "params":{"name":"agora","arguments":{}}
}'

# Calculation
curl -s http://127.0.0.1:8080/mcp -H 'content-type: application/json' -d '{
  "jsonrpc":"2.0","id":6,"method":"tools/call",
  "params":{"name":"calcular","arguments":{"expressao":"(2+3)*4"}}
}'

Cursor / Zed / 기타 MCP 클라이언트와의 통합

편집기 설정에 추가합니다 (예: ~/.cursor/settings.json):

{
  "mcpServers": {
    "rag-mcp": {
      "url": "http://127.0.0.1:8080/mcp"
    }
  }
}

클라이언트는 다음을 수행합니다:

  1. initialize 호출.

  2. 도구 목록 조회 (tools/list).

  3. 필요에 따라 ask_agent 사용 또는 도구 직접 호출.

RAG 작동 방식

  1. 인덱싱: 시작 시 SimpleDirectoryReader가 data/를 읽고 VectorStoreIndex가 text-embedding-3-small로 임베딩을 생성합니다.

  2. 쿼리: as_query_engine이 가장 유사한 3개 청크를 검색하고 LLM이 답변을 종합합니다.

  3. 업데이트: 다시 인덱싱하려면 data/에서 파일을 추가/제거하고 컨테이너를 재시작합니다.

에이전트 작동 방식 (LangGraph)

  • agent 노드는 bind_tools(TOOLS)로 LLM을 호출합니다.

  • tools_condition이 결정합니다: 모델이 함수 호출을 요청하면 tools 노드로 이동하고, 그렇지 않으면 종료합니다.

  • tools 노드는 도구를 실행하고 agent로 돌아가며, agent가 최종 응답을 생성합니다.

흐름:

START → agent → (tools?) → tools → agent → END

확장

새 도구 추가

app.py에 다음을 추가합니다:

@tool
def my_tool(param1: str, param2: int = 0) -> str:
    """Clear description of what the tool does."""
    # logic
    return "result"

그런 다음:

TOOLS.append(my_tool)

서버를 재시작합니다. 도구가 tools/list에 자동으로 나타납니다.

모델 변경

.env에서 LLM_MODEL을 변경합니다:

LLM_MODEL=gpt-4o

또는 app.py에서 ChatOpenAI와 임베딩을 교체하여 다른 제공자(예: Anthropic, Groq)를 사용할 수 있습니다.

벡터 스토어 변경

VectorStoreIndex를 영구 스토어(Chroma, Pinecone, Weaviate 등)로 교체합니다:

from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

client = chromadb.PersistentClient(path="./chroma")
collection = client.get_or_create_collection("rag")
vector_store = ChromaVectorStore(chroma_collection=collection)
_index = VectorStoreIndex.from_documents(docs, vector_store=vector_store)

보안 및 모범 사례

  • 서버를 인증 없이 인터넷에 직접 노출하지 마십시오.

  • MCP 클라이언트가 같은 호스트에 있으면 내부 네트워크(Docker)를 사용하세요.

  • 사용자 정의 도구에서 입력을 검증하세요 (특히 DB나 외부 API에 접근하는 경우).

  • 프로덕션 환경에서는 다음을 추가하세요:

    • 속도 제한(rate limiting).

    • 구조화된 로깅.

    • 메트릭(Prometheus, OpenTelemetry).

문제 해결

ModuleNotFoundError

  • requirements.txt를 설치했는지 확인하세요.

  • Docker에서는 docker compose build --no-cache를 실행하세요.

잘못된 OPENAI_API_KEY

  • docker compose exec mcp env | grep OPENAI로 키를 확인하세요.

  • 로컬 테스트: curl https://api.openai.com/v1/models -H "Authorization: Bearer $OPENAI_API_KEY".

RAG가 문서를 찾지 못함

  • data/에 유효한 파일(.md, .txt 등)이 있는지 확인하세요.

  • 로그 확인: docker compose logs mcp.

  • 컨테이너를 재시작하여 다시 인덱싱하세요.

calcular 오류

  • 이 도구는 간단한 산술 표현식만 허용합니다.

  • 변수, 함수, 복잡한 Python 구문은 피하세요.

파일

app.py

"""MCP Server (JSON-RPC) + LangGraph + LlamaIndex RAG."""
from __future__ import annotations

import ast
import operator as op
import os
from datetime import datetime, timezone
from typing import Annotated, TypedDict

from flask import Flask, jsonify, request
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI as LlamaLLM

DATA_DIR = os.getenv("DATA_DIR", "./data")
MODEL = os.getenv("LLM_MODEL", "gpt-4o-mini")

# --- RAG: index ./data once on process startup ---
Settings.llm = LlamaLLM(model=MODEL)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
_index = VectorStoreIndex.from_documents(SimpleDirectoryReader(DATA_DIR).load_data())
_qe = _index.as_query_engine(similarity_top_k=3)


# --- Function calling: each @tool becomes JSON schema for LLM and MCP ---
@tool
def rag_search(query: str) -> str:
    """Search facts in local base via RAG (LlamaIndex). Use for policies, products, and docs."""
    return str(_qe.query(query))


@tool
def agora() -> str:
    """Returns current UTC datetime (ISO-8601)."""
    return datetime.now(timezone.utc).isoformat()


@tool
def calcular(expressao: str) -> str:
    """Evaluates safe arithmetic. Examples: (2+3)*4, 10/2, 2**8."""
    ops = {
        ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv,
        ast.Mod: op.mod, ast.Pow: op.pow, ast.USub: op.neg,
    }

    def _eval(n):
        if isinstance(n, ast.Expression):
            return _eval(n.body)
        if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)):
            return n.value
        if isinstance(n, ast.BinOp) and type(n.op) in ops:
            return ops[type(n.op)](_eval(n.left), _eval(n.right))
        if isinstance(n, ast.UnaryOp) and type(n.op) in ops:
            return ops[type(n.op)](_eval(n.operand))
        raise ValueError("invalid expression")

    return str(_eval(ast.parse(expressao, mode="eval")))


TOOLS = [rag_search, agora, calcular]


# --- LangGraph: agent ↔ tools until model stops requesting function calls ---
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


llm = ChatOpenAI(model=MODEL, temperature=0).bind_tools(TOOLS)


def agent_node(state: State) -> dict:
    sys = SystemMessage(content="MCP assistant. Use tools when needed. Respond in English.")
    return {"messages": [llm.invoke([sys, *state["messages"]])]}


_g = StateGraph(State)
_g.add_node("agent", agent_node)
_g.add_node("tools", ToolNode(TOOLS))
_g.add_edge(START, "agent")
_g.add_conditional_edges("agent", tools_condition)  # tools or END
_g.add_edge("tools", "agent")
GRAPH = _g.compile()


def _schema(t) -> dict:
    """Converts LangChain tool to MCP inputSchema."""
    s = t.args_schema.model_json_schema() if t.args_schema else {"type": "object"}
    s.pop("title", None)
    return s


MCP_TOOLS = [
    {"name": t.name, "description": t.description, "inputSchema": _schema(t)}
    for t in TOOLS
] + [{
    "name": "ask_agent",
    "description": "LangGraph agent with RAG + function calling. Pass the user question.",
    "inputSchema": {
        "type": "object",
        "properties": {"question": {"type": "string"}},
        "required": ["question"],
    },
}]


def _run(name: str, args: dict) -> str:
    if name == "ask_agent":
        out = GRAPH.invoke({"messages": [HumanMessage(content=args.get("question", ""))]})
        return str(out["messages"][-1].content)
    fn = {t.name: t for t in TOOLS}.get(name)
    if not fn:
        raise ValueError(f"unknown tool: {name}")
    return str(fn.invoke(args or {}))


# --- Flask: HTTP transport for MCP (JSON-RPC 2.0) ---
app = Flask(__name__)


@app.post("/mcp")
def mcp():
    body = request.get_json(force=True) or {}
    method, rid, params = body.get("method"), body.get("id"), body.get("params") or {}

    if method == "initialize":
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "rag-mcp", "version": "1.0.0"},
        }})
    if method == "tools/list":
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": {"tools": MCP_TOOLS}})
    if method == "tools/call":
        try:
            text = _run(params.get("name"), params.get("arguments") or {})
            result = {"content": [{"type": "text", "text": text}]}
        except Exception as e:
            result = {"content": [{"type": "text", "text": str(e)}], "isError": True}
        return jsonify({"jsonrpc": "2.0", "id": rid, "result": result})
    if method == "notifications/initialized" or rid is None:
        return ("", 204)
    return jsonify({"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": method}}), 400


@app.get("/health")
def health():
    return {"ok": True}


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 8080)))

requirements.txt

flask>=3.0
langgraph>=0.2
langchain-core>=0.3
langchain-openai>=0.2
llama-index>=0.12
llama-index-llms-openai
llama-index-embeddings-openai

Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
COPY data ./data
ENV PORT=8080 DATA_DIR=/app/data
EXPOSE 8080
CMD ["python", "app.py"]

docker-compose.yml

services:
  mcp:
    build: .
    ports: ["8080:8080"]
    env_file: .env
    environment:
      PORT: "8080"
      DATA_DIR: /app/data
      LLM_MODEL: gpt-4o-mini
    volumes:
      - ./data:/app/data:ro
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/health')"]
      interval: 15s
      retries: 5

data/kb.md

# Acme Corp
Support SLA: 4 hours during business hours (UTC-3).
Pro Plan costs USD 49/month and includes 10k RAG queries/day.
P1 incidents must be opened in #sre channel.

.env

OPENAI_API_KEY=sk-...
LLM_MODEL=gpt-4o-mini

라이선스

MIT.

This server cannot be deployed

One endpoint for every MCP toolRayrun runs every MCP server behind one URL and holds the credentials. Each call is decided per client, tool, and argument, then recorded.ray.runAd

Maintenance

ActivityMaintained
ResponsivenessNo issues

Resources

Related MCP Connectors

Related MCP Servers