de en es ja ko ru zh

rag-mcp

Name: rag-mcp
Author: johnylira

by johnylira

Overview Schema Source Related Servers Discussions Admin

Python

Hybrid

RAG-MCP Server

基于 Flask、LangGraph 和 LlamaIndex 构建的 Model Context Protocol (MCP) 服务器，用于 RAG。设计简洁、功能完备且易于扩展。

概述

基于 HTTP 的 MCP：POST /mcp，使用 JSON-RPC 2.0。
LangGraph：通过函数调用编排智能体。
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. 克隆 / 创建项目

创建一个空目录并粘贴项目文件（参见文件部分）。

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

客户端将：

调用 initialize。
列出工具（tools/list）。
使用 ask_agent 或根据需要直接调用工具。

RAG 工作原理

索引：启动时，SimpleDirectoryReader 读取 data/，VectorStoreIndex 使用 text-embedding-3-small 创建嵌入。
查询：as_query_engine 检索最相似的 3 个文本块，LLM 综合生成答案。
更新：要重新索引，请在 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）。
在自定义工具中验证输入（尤其是在访问数据库或外部 API 时）。
对于生产环境，请添加：
- 速率限制。
- 结构化日志。
- 指标（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

GitHub Repository

Need Help?

Related Servers

Related MCP Connectors

docs2mcpOAuth
com.docs2mcp
Query your own PDFs and documents from any MCP client. Every answer cites the page it came from.
RAG Systems Search
1
saagarpatel.dev Portfolio A
io.github.saagpatel
Agent-native MCP server over the public saagarpatel.dev corpus. Read-only, stateless.
Search Documentation Access
82
CustomGPT.ai
ai.customgpt.mcp
The CustomGPT.ai MCP server is a fully managed, RAG-powered endpoint that connects large language models with private knowledge bases and external data sources. It provides tools for retrieval-augmented generation queries (send_message), data ingestion (upload_file), and source listing, enabling AI agents to query private documents like PDFs with high accuracy and real-time citations.
RAG Systems Knowledge & Memory Search
2
agentready-mcp A
io.github.AshutoshRaj97
Query any docs site via MCP. Submit a URL, ask questions, get cited answers.
Documentation Access Search
822

Related MCP Servers

LangChain MCP Demo
RAG Systems Search Text Summarization
shafeequealipt-dotcom
F
license
Not graded
quality
D
maintenance
Enables grounding AI responses in a local document corpus by exposing MCP tools to list, search, and summarize documents, and generating answers using OpenAI.
Updated 3 months ago (2026-06-04 05:39 UTC)
-
DocAgent-MCP
RAG Systems Search Text Summarization
wuliuke
F
license
Not graded
quality
C
maintenance
Enables local document question-answering and retrieval via MCP, supporting multi-turn conversation, intent recognition, and tools for document search, Q&A, and summarization.
Updated 2 months ago (2026-07-14 07:45 UTC)
5
-
AI Document Assistant MCP
RAG Systems AI & Machine Learning Search
yadu0323
F
license
Not graded
quality
B
maintenance
Enables document Q&A, summarization, keyword extraction, and Wikipedia lookup through MCP tools, using RAG with FAISS and Ollama.
Updated a month ago (2026-08-19 17:16 UTC)
-
Agentic RAG Assistant
RAG Systems AI & Machine Learning Autonomous Agents
snehaharurkar
F
license
Not graded
quality
C
maintenance
Enables querying internal documents via a FastAPI REST API and MCP server, using retrieval-augmented generation and an agentic loop that can invoke tools like document search and calculations.
Updated a month ago (2026-08-20 07:01 UTC)
-

RAG-MCP Server

概述

架构

前提条件

项目结构

配置

1. 克隆 / 创建项目

2. 环境变量

3. RAG 数据

运行

Docker（推荐）

本地运行（不使用 Docker）

端点

POST /mcp（JSON-RPC 2.0）

initialize

tools/list

tools/call

GET /health

可用工具

直接使用示例

与 Cursor / Zed / 其他 MCP 客户端集成

RAG 工作原理

智能体工作原理（LangGraph）

扩展

添加新工具

更换模型

更换向量存储

安全与最佳实践

故障排除

ModuleNotFoundError

OPENAI_API_KEY 无效

RAG 找不到文档

calcular 中的错误

文件

app.py

requirements.txt

Dockerfile

docker-compose.yml

data/kb.md

.env

许可证