Building an Advanced RAG System (MCP+RAG+WEB_SEARCH)

The frontend leverages Streamlit with extensive custom CSS for a modern, responsive interface:

# Modern CSS styling with gradient backgrounds
import streamlit as st

st.set_page_config(
    page_title="IntelliSearch RAG",
    page_icon="🚀",
    layout="wide",
    initial_sidebar_state="expanded"
)

# Hero section with gradient background
st.markdown("""
<div class="hero-header">
    <h1 class="hero-title">🚀 IntelliSearch RAG</h1>
    <p class="hero-subtitle">Advanced AI-Powered Search with RAG</p>
</div>
""", unsafe_allow_html=True)

# Enhanced tabs for results display
tab1, tab2, tab3, tab4 = st.tabs([
    "🌐 Web Results", 
    "🧠 AI Analysis", 
    "📑 Source Documents", 
    "📈 Insights"
])

The MCP client handles communication between the frontend and backend services:

class LangchainMCPClient:
    def __init__(self, mcp_server_url="http://localhost:8000"):
        self.llm = ChatOllama(
            model="llama2-70b",
            temperature=0.7,
            streaming=False
        )
        
        # Server configuration with timeout and retry settings
        server_config = {
            "default": {
                "url": f"{mcp_server_url}/sse",
                "transport": "sse",
                "options": {
                    "timeout": 30.0,
                    "retry_connect": True,
                    "max_retries": 3
                }
            }
        }
        
        self.mcp_client = MultiServerMCPClient(server_config)

    async def process_message(self, user_input: str):
        # Process query through MCP server
        tool = self.tools[0]
        result = await tool.coroutine(user_input)
        return result

High-quality web search using the Exa API for relevant, up-to-date information:

async def search_web(query: str, num_results: int = 5):
    # Initialize Exa client
    exa = Exa(api_key=os.getenv("EXA_API_KEY"))
    
    # Perform search with content and summaries
    search_results = exa.search_and_contents(
        query,
        num_results=num_results,
        summary={"query": "Main points and key takeaways"}
    )
    
    # Format results for display
    formatted_results = format_search_results(search_results)
    return formatted_results, search_results.results

async def get_web_content(url: str):
    # Extract content using BeautifulSoup
    headers = {"User-Agent": USER_AGENT}
    response = requests.get(url, headers=headers, timeout=30)
    
    soup = BeautifulSoup(response.text, 'html.parser')
    
    # Remove script and style elements
    for script in soup(["script", "style"]):
        script.decompose()
    
    text = soup.get_text(separator='\n', strip=True)
    return [Document(page_content=text, metadata={"source": url})]

The RAG system creates embeddings and performs similarity search for relevant context:

async def create_rag_from_documents(documents: List[Document]):
    # Initialize Ollama embeddings
    embeddings = OllamaEmbeddings(
        model="mxbai-embed-large:latest",
        base_url="http://localhost:11434"
    )
    
    # Split documents into chunks
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=2000,
        chunk_overlap=200,
        length_function=len,
    )
    split_documents = text_splitter.split_documents(documents)
    
    # Create FAISS vector store
    vectorstore = FAISS.from_documents(
        documents=split_documents, 
        embedding=embeddings
    )
    return vectorstore

async def search_rag(query: str, vectorstore: FAISS, k: int = 5):
    # Perform similarity search
    results = vectorstore.similarity_search(query, k=k)
    return results

The MCP server orchestrates the entire search and analysis pipeline:

# Initialize FastMCP server
mcp = FastMCP(
    name="web_search_rag",
    version="1.0.0",
    description="Advanced web search with RAG integration",
    host="localhost",
    type="sse",
    port=8000,
    timeout=30,
    keep_alive=True
)

@mcp.tool()
async def search_and_analyze(
    query: str,
    num_results: int = 5,
    rag_results: int = 3
):
    # Perform web search
    formatted_results, raw_results = await search.search_web(query, num_results)
    
    # Extract URLs and create RAG system
    urls = [result.url for result in raw_results]
    vectorstore = await rag.create_rag(urls)
    rag_results = await rag.search_rag(query, vectorstore, k=rag_results)
    
    # Format and return response
    return {
        "search_results": formatted_results,
        "rag_analysis": [
            {
                "content": doc.page_content,
                "metadata": {"source": doc.metadata.get("source")}
            } for doc in rag_results
        ]
    }