Skip to main content
Glama
deenesjakoruzh
johnylira

rag-mcp

by johnylira
OverviewSchemaSourceRelated ServersDiscussionsAdmin
Python
Hybrid

RAG-MCP Server

Model Context Protocol (MCP)-Server, erstellt mit Flask, LangGraph und LlamaIndex für RAG. Entwickelt, um minimal, funktional und einfach erweiterbar zu sein.

Überblick

  • MCP über HTTP: POST /mcp mit JSON-RPC 2.0.

  • LangGraph: orchestriert den Agenten mit Funktionsaufrufen.

  • LlamaIndex: indexiert lokale Dokumente und beantwortet Fragen über RAG.

  • Docker: docker compose up --build und du bist fertig.

Related MCP server: DocAgent-MCP

Architektur

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

  • Der Client listet Tools (tools/list) auf und ruft sie auf (tools/call).

  • Das ask_agent-Tool löst den LangGraph-Graphen aus; das LLM entscheidet, wann Funktionsaufrufe verwendet werden.

  • Andere Tools werden direkt von MCP aufgerufen.

Voraussetzungen

  • Docker und Docker Compose (oder lokal Python 3.11+).

  • OPENAI_API_KEY (für LLM und Embeddings).

Projektstruktur

.
├── 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

Konfiguration

1. Projekt klonen / erstellen

Erstelle ein leeres Verzeichnis und füge die Projektdateien ein (siehe Abschnitt Dateien).

2. Umgebungsvariablen

Erstelle .env im Stammverzeichnis:

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

Unterstützte Variablen:

Variable

Standard

Beschreibung

OPENAI_API_KEY

(erforderlich)

OpenAI-API-Schlüssel.

LLM_MODEL

gpt-4o-mini

Modell für LLM und Embeddings.

DATA_DIR

/app/data

Verzeichnis mit Dokumenten für RAG.

PORT

8080

Server-Port.

3. RAG-Daten

Lege Dokumente in data/ ab (z.B. kb.md, policies.md, manuals/). Der Server indexiert beim Start alles.

Minimales Beispiel (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.

Ausführen

Docker (empfohlen)

docker compose up --build

Der Server läuft unter http://127.0.0.1:8080.

Lokal (ohne Docker)

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

Endpunkte

POST /mcp (JSON-RPC 2.0)

MCP verwendet drei Hauptmethoden:

initialize

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

Antwort:

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

tools/list

Listet alle verfügbaren Tools auf (einschließlich 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

Ruft ein Tool auf:

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?"}
  }
}'

Antwort:

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

Einfacher Health-Check:

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

Verfügbare Tools

Tool

Beschreibung

InputSchema

ask_agent

LangGraph-Agent mit RAG + Funktionsaufrufen.

{"question": "string"}

rag_search

Sucht Fakten in der Basis über RAG (LlamaIndex).

{"query": "string"}

agora

Gibt die aktuelle UTC-Zeit zurück (ISO-8601).

{}

calcular

Bewertet einen sicheren arithmetischen Ausdruck.

{"expressao": "string"}

Direkte Verwendungsbeispiele

# 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"}}
}'

Integration mit Cursor / Zed / Anderen MCP-Clients

Füge dies zu deiner Editor-Konfiguration hinzu (z.B. ~/.cursor/settings.json):

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

Der Client wird:

  1. initialize aufrufen.

  2. Tools auflisten (tools/list).

  3. ask_agent verwenden oder Tools direkt nach Bedarf aufrufen.

So funktioniert RAG

  1. Indexierung: Beim Start liest SimpleDirectoryReader data/ und VectorStoreIndex erstellt Embeddings mit text-embedding-3-small.

  2. Abfrage: as_query_engine ruft die 3 ähnlichsten Chunks ab und das LLM synthetisiert die Antwort.

  3. Aktualisierung: Zum Neuindizieren Dateien in data/ hinzufügen/entfernen und den Container neu starten.

So funktioniert der Agent (LangGraph)

  • Der agent-Knoten ruft das LLM mit bind_tools(TOOLS) auf.

  • tools_condition entscheidet: Wenn das Modell Funktionsaufrufe anfordert, geht es zum tools-Knoten; andernfalls endet es.

  • Der tools-Knoten führt das Tool aus und kehrt zu agent zurück, das die endgültige Antwort generiert.

Ablauf:

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

Erweiterung

Neues Tool hinzufügen

In app.py hinzufügen:

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

Dann:

TOOLS.append(my_tool)

Starte den Server neu. Das Tool erscheint automatisch in tools/list.

Modell ändern

Ändere LLM_MODEL in .env:

LLM_MODEL=gpt-4o

Oder verwende einen anderen Anbieter (z.B. Anthropic, Groq), indem du ChatOpenAI und Embeddings in app.py ersetzt.

Vektor-Speicher ändern

Ersetze VectorStoreIndex durch einen persistenten Speicher (Chroma, Pinecone, Weaviate usw.):

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)

Sicherheit und Best Practices

  • Setze den Server nicht ohne Authentifizierung direkt dem Internet aus.

  • Verwende internes Netzwerk (Docker), wenn der MCP-Client auf demselben Host läuft.

  • Validiere Eingaben in benutzerdefinierten Tools (insbesondere bei Zugriff auf Datenbanken oder externe APIs).

  • Für die Produktion füge hinzu:

    • Rate-Limiting.

    • Strukturierte Protokollierung.

    • Metriken (Prometheus, OpenTelemetry).

Fehlerbehebung

ModuleNotFoundError

  • Überprüfe, ob du requirements.txt installiert hast.

  • In Docker führe docker compose build --no-cache aus.

Ungültiger OPENAI_API_KEY

  • Bestätige den Schlüssel mit docker compose exec mcp env | grep OPENAI.

  • Teste lokal: curl https://api.openai.com/v1/models -H "Authorization: Bearer $OPENAI_API_KEY".

RAG findet keine Dokumente

  • Überprüfe, ob data/ gültige Dateien enthält (.md, .txt usw.).

  • Überprüfe die Logs: docker compose logs mcp.

  • Indiziere neu, indem du den Container neu startest.

Fehler in calcular

  • Das Tool akzeptiert nur einfache arithmetische Ausdrücke.

  • Vermeide Variablen, Funktionen oder komplexe Python-Syntax.

Dateien

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

Lizenz

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