Skip to main content
Glama
deenesjakoruzh
johnylira

rag-mcp

by johnylira
OverviewSchemaSourceRelated ServersDiscussionsAdmin
Python
Hybrid

Servidor RAG-MCP

Servidor Model Context Protocol (MCP) construido con Flask, LangGraph y LlamaIndex para RAG. Diseñado para ser mínimo, funcional y fácil de extender.

Resumen

  • MCP sobre HTTP: POST /mcp con JSON-RPC 2.0.

  • LangGraph: orquesta el agente con llamada a funciones.

  • LlamaIndex: indexa documentos locales y responde mediante RAG.

  • Docker: docker compose up --build y listo.

Related MCP server: DocAgent-MCP

Arquitectura

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

  • El cliente lista herramientas (tools/list) y las llama (tools/call).

  • La herramienta ask_agent activa el grafo de LangGraph; el LLM decide cuándo usar la llamada a funciones.

  • Otras herramientas se llaman directamente por MCP.

Requisitos previos

  • Docker y Docker Compose (o Python 3.11+ localmente).

  • OPENAI_API_KEY (para LLM y embeddings).

Estructura del proyecto

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

Configuración

1. Clonar / Crear el proyecto

Crea un directorio vacío y pega los archivos del proyecto (consulta la sección Archivos).

2. Variables de entorno

Crea .env en la raíz:

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

Variables admitidas:

Variable

Valor por defecto

Descripción

OPENAI_API_KEY

(obligatorio)

Clave de API de OpenAI.

LLM_MODEL

gpt-4o-mini

Modelo para LLM y embeddings.

DATA_DIR

/app/data

Directorio con documentos para RAG.

PORT

8080

Puerto del servidor.

3. Datos RAG

Coloca documentos en data/ (por ejemplo, kb.md, policies.md, manuals/). El servidor indexa todo al iniciar.

Ejemplo mínimo (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.

Ejecución

Docker (recomendado)

docker compose up --build

El servidor se ejecuta en http://127.0.0.1:8080.

Local (sin Docker)

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

Endpoints

POST /mcp (JSON-RPC 2.0)

MCP utiliza tres métodos principales:

initialize

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

Respuesta:

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

tools/list

Lista todas las herramientas disponibles (incluyendo 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

Llama a una herramienta:

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

Respuesta:

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

Comprobación de salud simple:

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

Herramientas disponibles

Herramienta

Descripción

InputSchema

ask_agent

Agente LangGraph con RAG + llamada a funciones.

{"question": "string"}

rag_search

Busca hechos en la base mediante RAG (LlamaIndex).

{"query": "string"}

agora

Devuelve la fecha y hora UTC actual (ISO-8601).

{}

calcular

Evalúa una expresión aritmética segura.

{"expressao": "string"}

Ejemplos de uso directo

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

Integración con Cursor / Zed / Otros clientes MCP

Añade a la configuración de tu editor (por ejemplo, ~/.cursor/settings.json):

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

El cliente:

  1. Llamará a initialize.

  2. Listará las herramientas (tools/list).

  3. Usará ask_agent o llamará a las herramientas directamente según sea necesario.

Cómo funciona RAG

  1. Indexación: Al iniciar, SimpleDirectoryReader lee data/ y VectorStoreIndex crea embeddings con text-embedding-3-small.

  2. Consulta: as_query_engine recupera los 3 fragmentos más similares y el LLM sintetiza la respuesta.

  3. Actualización: Para reindexar, añade/elimina archivos en data/ y reinicia el contenedor.

Cómo funciona el agente (LangGraph)

  • El nodo agent invoca al LLM con bind_tools(TOOLS).

  • tools_condition decide: si el modelo solicita llamada a funciones, va al nodo tools; de lo contrario, termina.

  • El nodo tools ejecuta la herramienta y vuelve a agent, que genera la respuesta final.

Flujo:

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

Extensión

Añadir una nueva herramienta

En app.py, añade:

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

Luego:

TOOLS.append(my_tool)

Reinicia el servidor. La herramienta aparece automáticamente en tools/list.

Cambiar el modelo

Cambia LLM_MODEL en .env:

LLM_MODEL=gpt-4o

O usa otro proveedor (por ejemplo, Anthropic, Groq) reemplazando ChatOpenAI y los embeddings en app.py.

Cambiar el almacén de vectores

Reemplaza VectorStoreIndex con un almacén persistente (Chroma, Pinecone, Weaviate, etc.):

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)

Seguridad y mejores prácticas

  • No expongas el servidor directamente a Internet sin autenticación.

  • Usa red interna (Docker) si el cliente MCP está en el mismo host.

  • Valida las entradas en herramientas personalizadas (especialmente si acceden a bases de datos o APIs externas).

  • Para producción, añade:

    • Limitación de velocidad.

    • Registro estructurado.

    • Métricas (Prometheus, OpenTelemetry).

Solución de problemas

ModuleNotFoundError

  • Verifica que instalaste requirements.txt.

  • En Docker, ejecuta docker compose build --no-cache.

Clave OPENAI_API_KEY no válida

  • Confirma la clave en docker compose exec mcp env | grep OPENAI.

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

RAG no encuentra documentos

  • Verifica que data/ tenga archivos válidos (.md, .txt, etc.).

  • Revisa los registros: docker compose logs mcp.

  • Reindexa reiniciando el contenedor.

Error en calcular

  • La herramienta solo acepta expresiones aritméticas simples.

  • Evita variables, funciones o sintaxis compleja de Python.

Archivos

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

Licencia

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