Modular RAG MCP Server

Modular RAG MCP Server

Skill by ara.so — MCP Skills collection.

Expert skill for deploying, configuring, and extending the Modular RAG MCP Server — a pluggable, observable RAG (Retrieval-Augmented Generation) system that exposes tools via Model Context Protocol for AI assistants like Claude Desktop and GitHub Copilot.

What This Project Does

The Modular RAG MCP Server is a complete RAG pipeline featuring:

• Ingestion Pipeline: PDF → Markdown → Chunking → Embedding → Vector Store (with multimodal image captioning)
• Hybrid Search: Dense vectors (semantic) + Sparse BM25 (exact match) + RRF fusion + optional reranking
• MCP Protocol: Standard MCP server exposing query_knowledge_hub, list_collections, get_document_summary tools
• Dashboard: Streamlit-based management UI with 6 pages (overview, data browser, ingestion tracking, query tracking, evaluation)
• Evaluation Framework: Ragas + custom metrics for regression testing
• Full Observability: White-box tracing of ingestion and query pipelines

Key Architecture: Every core component (LLM, Embedding, Reranker, Splitter, VectorStore, Evaluator) is pluggable via abstract interfaces. Switch backends through configuration without code changes.

Installation

Prerequisites

• Python 3.9+
• VS Code with GitHub Copilot or Claude Desktop
• API keys for your chosen providers (OpenAI, Anthropic, Cohere, etc.)

Quick Setup with Setup Skill

The project includes a Setup Skill that automates the entire configuration:

# Clone the repository
git clone https://github.com/jerry-ai-dev/MODULAR-RAG-MCP-SERVER.git
cd MODULAR-RAG-MCP-SERVER

# In VS Code with Copilot/Claude, type in chat:
setup

The Setup Skill will:

1. Ask you to select providers (OpenAI, Anthropic, Cohere, etc.)
2. Configure API keys
3. Install dependencies
4. Generate configuration files
5. Launch the dashboard

Manual Setup

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Copy and configure environment variables
cp .env.example .env
# Edit .env with your API keys

Configuration

Main Configuration File (src/core/config.py)

The system uses a centralized configuration approach. Key settings:

from src.core.config import get_config

config = get_config()

# Access configuration
llm_provider = config.llm.provider  # "openai", "anthropic", etc.
embedding_provider = config.embedding.provider
vector_store_type = config.vector_store.type  # "qdrant", "chroma", etc.

Environment Variables

Create .env file with required keys:

# LLM Provider
OPENAI_API_KEY=your_openai_key_here
ANTHROPIC_API_KEY=your_anthropic_key_here

# Embedding Provider
COHERE_API_KEY=your_cohere_key_here

# Reranker (optional)
JINA_API_KEY=your_jina_key_here

# Vector Store (if using cloud)
QDRANT_URL=your_qdrant_url
QDRANT_API_KEY=your_qdrant_key

Provider Configuration

Edit src/core/config.py to set default providers:

class LLMConfig:
    provider: str = "openai"  # or "anthropic", "cohere"
    model: str = "gpt-4"
    temperature: float = 0.7
    max_tokens: int = 2048

class EmbeddingConfig:
    provider: str = "openai"  # or "cohere", "huggingface"
    model: str = "text-embedding-3-small"
    dimension: int = 1536

class RerankerConfig:
    enabled: bool = True
    provider: str = "cohere"  # or "jina", "cross-encoder"
    model: str = "rerank-english-v3.0"
    top_k: int = 5

Key Components and API

1. Ingestion Pipeline

Ingest documents into the knowledge base:

from src.ingestion.pipeline import IngestionPipeline
from src.core.config import get_config

config = get_config()
pipeline = IngestionPipeline(config)

# Ingest a PDF document
result = pipeline.ingest_document(
    file_path="path/to/document.pdf",
    collection_name="my_collection",
    metadata={"source": "internal_docs", "version": "1.0"}
)

print(f"Ingested {result['chunks_created']} chunks")
print(f"Ingestion ID: {result['ingestion_id']}")

2. Hybrid Search and Query

Query the knowledge base with hybrid search:

from src.retrieval.hybrid_search import HybridSearchRetriever
from src.core.config import get_config

config = get_config()
retriever = HybridSearchRetriever(config)

# Perform hybrid search
results = retriever.retrieve(
    query="How does the authentication system work?",
    collection_name="my_collection",
    top_k=10,  # Initial retrieval
    rerank_top_k=5  # After reranking
)

for idx, result in enumerate(results):
    print(f"{idx+1}. Score: {result.score:.4f}")
    print(f"   Text: {result.text[:100]}...")
    print(f"   Metadata: {result.metadata}")

3. MCP Server Integration

The MCP server exposes tools for AI assistants. Start the server:

# Run MCP server (usually configured in Claude Desktop config)
python src/mcp/server.py

Configure in Claude Desktop (claude_desktop_config.json):

{
  "mcpServers": {
    "rag-knowledge-hub": {
      "command": "python",
      "args": ["/path/to/project/src/mcp/server.py"],
      "env": {
        "PYTHONPATH": "/path/to/project"
      }
    }
  }
}

Available MCP Tools:

1. query_knowledge_hub: Query the RAG system

   # When Claude calls this tool:
   {
     "query": "What are the deployment requirements?",
     "collection_name": "my_collection",
     "top_k": 5
   }
   

2. list_collections: List all available collections

   # Returns: ["collection1", "collection2", ...]
   

3. get_document_summary: Get summary of a specific document

   {
     "document_id": "doc_123",
     "collection_name": "my_collection"
   }
   

4. Dashboard

Launch the Streamlit dashboard:

streamlit run src/dashboard/app.py

Dashboard pages:

• Overview: System status, collection stats, recent activity
• Data Browser: Browse and search ingested documents
• Ingestion Management: Upload new documents, view ingestion history
• Ingestion Tracking: Monitor ingestion pipeline steps
• Query Tracking: Analyze query performance and results
• Evaluation Panel: Run evaluations with Ragas metrics

5. Evaluation with Ragas

Evaluate RAG performance:

from src.evaluation.evaluator import RAGEvaluator
from src.core.config import get_config

config = get_config()
evaluator = RAGEvaluator(config)

# Prepare test dataset
test_cases = [
    {
        "query": "What is the API rate limit?",
        "expected_answer": "The API rate limit is 1000 requests per hour.",
        "ground_truth_context": ["Rate limits are set to 1000 req/hour..."]
    },
    # ... more test cases
]

# Run evaluation
results = evaluator.evaluate(
    test_cases=test_cases,
    collection_name="my_collection",
    metrics=["faithfulness", "answer_relevancy", "context_precision"]
)

print(f"Average Faithfulness: {results['faithfulness']:.3f}")
print(f"Average Answer Relevancy: {results['answer_relevancy']:.3f}")

Common Patterns

Switching Embedding Providers

To switch from OpenAI to Cohere embeddings:

# In src/core/config.py
class EmbeddingConfig:
    provider: str = "cohere"  # Changed from "openai"
    model: str = "embed-english-v3.0"
    dimension: int = 1024  # Cohere dimension

Or programmatically:

from src.core.config import get_config

config = get_config()
config.embedding.provider = "cohere"
config.embedding.model = "embed-english-v3.0"
config.embedding.dimension = 1024

Adding Custom Chunking Strategy

Implement a custom text splitter:

from src.ingestion.splitters.base import BaseSplitter
from typing import List

class CustomSplitter(BaseSplitter):
    def __init__(self, chunk_size: int = 500, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def split(self, text: str, metadata: dict = None) -> List[dict]:
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            chunk_text = text[start:end]
            chunks.append({
                "text": chunk_text,
                "metadata": {
                    **(metadata or {}),
                    "chunk_index": len(chunks),
                    "start_char": start
                }
            })
            start += self.chunk_size - self.overlap
        return chunks

# Register and use
from src.ingestion.pipeline import IngestionPipeline

pipeline = IngestionPipeline(config)
pipeline.splitter = CustomSplitter(chunk_size=300, overlap=30)

Implementing Custom Reranker

from src.retrieval.rerankers.base import BaseReranker
from typing import List

class CustomReranker(BaseReranker):
    def rerank(self, query: str, documents: List[dict], top_k: int = 5) -> List[dict]:
        # Custom reranking logic
        scored_docs = []
        for doc in documents:
            # Example: simple keyword matching score
            score = sum(1 for word in query.lower().split() 
                       if word in doc['text'].lower())
            scored_docs.append({**doc, 'rerank_score': score})
        
        # Sort by score and return top_k
        scored_docs.sort(key=lambda x: x['rerank_score'], reverse=True)
        return scored_docs[:top_k]

# Use in retriever
from src.retrieval.hybrid_search import HybridSearchRetriever

retriever = HybridSearchRetriever(config)
retriever.reranker = CustomReranker()

Multimodal Image Processing

The system supports image captioning in PDFs:

from src.ingestion.pipeline import IngestionPipeline

pipeline = IngestionPipeline(config)

# Enable image captioning
result = pipeline.ingest_document(
    file_path="document_with_images.pdf",
    collection_name="multimodal_docs",
    enable_image_captioning=True,  # Vision LLM generates descriptions
    metadata={"type": "technical_manual"}
)

# Images are converted to text descriptions and embedded with surrounding text

Batch Ingestion

Ingest multiple documents:

import os
from pathlib import Path

pipeline = IngestionPipeline(config)
docs_dir = Path("./documents")

results = []
for pdf_file in docs_dir.glob("*.pdf"):
    try:
        result = pipeline.ingest_document(
            file_path=str(pdf_file),
            collection_name="batch_collection",
            metadata={"filename": pdf_file.name}
        )
        results.append(result)
        print(f"✓ Ingested {pdf_file.name}")
    except Exception as e:
        print(f"✗ Failed {pdf_file.name}: {e}")

print(f"Total successful: {len(results)}")

Troubleshooting

MCP Server Not Connecting

Issue: Claude Desktop cannot connect to MCP server

Solution:

1. Check Claude Desktop config path (macOS: ~/Library/Application Support/Claude/claude_desktop_config.json)
2. Ensure Python path and project path are absolute
3. Verify environment variables are set in config:

   {
     "mcpServers": {
       "rag-knowledge-hub": {
         "command": "/usr/bin/python3",
         "args": ["/absolute/path/to/project/src/mcp/server.py"],
         "env": {
           "PYTHONPATH": "/absolute/path/to/project",
           "OPENAI_API_KEY": "sk-..."
         }
       }
     }
   }
   

4. Restart Claude Desktop completely

Poor Retrieval Results

Issue: Query returns irrelevant documents

Solutions:

1. Check chunking strategy: Smaller chunks for precise retrieval, larger for more context

   config.ingestion.chunk_size = 300  # Reduce for precision
   config.ingestion.chunk_overlap = 50
   

2. Enable reranking: Use cross-encoder or LLM reranker

   config.reranker.enabled = True
   config.reranker.provider = "cohere"
   config.reranker.top_k = 5
   

3. Adjust hybrid search weights:

   from src.retrieval.hybrid_search import HybridSearchRetriever
   
   retriever = HybridSearchRetriever(config)
   retriever.dense_weight = 0.7  # Semantic search
   retriever.sparse_weight = 0.3  # BM25 exact match
   

4. Use evaluation to iterate:

   # Create golden test set
   evaluator = RAGEvaluator(config)
   results = evaluator.evaluate(test_cases, collection_name="my_collection")
   # Adjust parameters based on metrics
   

Vector Store Connection Issues

Issue: Cannot connect to Qdrant/Chroma

Solution:

1. For Qdrant Cloud:

   # .env
   QDRANT_URL=https://your-cluster.qdrant.io
   QDRANT_API_KEY=your_api_key
   

2. For local Qdrant:

   # Start Qdrant with Docker
   docker run -p 6333:6333 qdrant/qdrant
   
   # In config
   QDRANT_URL=http://localhost:6333
   

3. For Chroma (local):

   # config.py
   class VectorStoreConfig:
       type: str = "chroma"
       persist_directory: str = "./chroma_db"
   

Out of Memory During Ingestion

Issue: Large PDFs cause OOM errors

Solutions:

1. Process in batches:

   # Increase chunk size, reduce batch size
   config.ingestion.chunk_size = 800
   config.ingestion.batch_size = 10  # Embed 10 chunks at a time
   

2. Use streaming for large documents:

   pipeline = IngestionPipeline(config)
   pipeline.process_streaming(
       file_path="large_document.pdf",
       collection_name="large_docs"
   )
   

API Rate Limits

Issue: Hitting provider rate limits

Solutions:

1. Implement retry with exponential backoff:

   config.llm.max_retries = 5
   config.llm.retry_delay = 2.0  # seconds
   

2. Use batch embedding APIs:

   # OpenAI allows batching up to 2048 texts
   config.embedding.batch_size = 100
   

3. Switch to providers with higher limits (e.g., Cohere for embeddings)

Advanced Usage

Custom RAG Pipeline

Build a custom RAG pipeline with specific components:

from src.core.config import get_config
from src.retrieval.hybrid_search import HybridSearchRetriever
from src.generation.generator import Generator
from src.evaluation.evaluator import RAGEvaluator

config = get_config()

# Custom retriever configuration
retriever = HybridSearchRetriever(config)
retriever.dense_weight = 0.6
retriever.sparse_weight = 0.4

# Custom generator
generator = Generator(config)
generator.system_prompt = "You are a helpful technical assistant..."

# Run custom RAG
def custom_rag_query(query: str, collection: str):
    # Retrieve
    contexts = retriever.retrieve(query, collection, top_k=5)
    
    # Generate
    response = generator.generate(
        query=query,
        contexts=[c.text for c in contexts],
        metadata=[c.metadata for c in contexts]
    )
    
    # Evaluate (optional)
    evaluator = RAGEvaluator(config)
    metrics = evaluator.evaluate_single(
        query=query,
        response=response,
        contexts=[c.text for c in contexts]
    )
    
    return {
        "response": response,
        "contexts": contexts,
        "metrics": metrics
    }

result = custom_rag_query("What are the system requirements?", "docs")
print(result["response"])

Integrating with Your Own Application

Use the RAG system as a library:

from src.rag_system import RAGSystem
from src.core.config import get_config

# Initialize
config = get_config()
rag = RAGSystem(config)

# In your FastAPI/Flask app
@app.post("/ask")
async def ask_question(query: str, collection: str = "default"):
    result = rag.query(
        query=query,
        collection_name=collection,
        top_k=5
    )
    return {
        "answer": result["response"],
        "sources": result["contexts"],
        "confidence": result["metrics"]["answer_relevancy"]
    }

Branch Strategy

• main: Clean, production-ready code (1 commit with latest complete code)
• dev: Full commit history showing development progression
• clean-start: Skeleton with Skills and DEV_SPEC, zero progress (for learning from scratch)

Choose branch based on your needs:

• Quick deployment → main
• Understanding the build process → dev
• Learning by building yourself → clean-start

Additional Resources

• DEV_SPEC.md: Complete architecture design and task breakdown
• Resume Writer Skill: Generate customized resume descriptions for this project
• QA Tester Skill: Automated testing across unit/integration/E2E layers
• Package Skill: Clean and package project for distribution

Use these skills in VS Code by typing the skill name in Copilot/Claude chat.