Generative AI Series

Hands on — Agentic RAG (3/3) — Agentic Integrated with MCP Server

This blog is a final part of the 3 part series on Agentic RAG. In this blog we will build a agentic RAG system, that integrates with a web search MCP. We will be building a MCP server from scratch.

Courtesy: ChatGPT

This blog is a final part of the 3 part series on Agentic RAG. In this blog we will extend the Agentic RAG application that we built in Part 2, and integrate that MCP server. We will also build a MCP server without any frameworks, to understand the complete end to end working of the Agentic AI application with MCP server.

Before we go any further, I strongly suggest you read the following blogs.

• Retrieval Augmented Generation (RAG)
• Retrieval Augmented Generation(RAG) — Chatbot for documents with LlamaIndex
• Retrieval Augmented Generation(RAG) — Chatbot for database with LlamaIndex (Text2SQL)
• Retrieval Augmented Generation(RAG) — Chatbot for Wikipedia with LlamaIndex
• Retrieval Augmented Generation(RAG) — Chatbot for Youtube with LlamaIndex

and don’t forget to read Part 1 of the blog Hands on — Agentic RAG (1/2), In that blog, I had shared my thoughts about Agentic RAG and how its different from Traditional RAG. In Part 2 Hands on — Agentic RAG (2/3) — Agentic ReRanking RAG we built a reranking RAG Agentic AI application. In this blog we will extend that code to integrate with a MCP Server.

Now lets hit the road, and get our hands dirty.

MCP Server Tutorial

Lets first build the web-search MCP Server. We will be building this using FastAPI, not MCP server SDK, to get a complete understanding. This is much easier if I had used FastMCP…but then you will find that in various other blogs/youtube…so no fun :-D, will also be a good learning experience. the MCP server we are going to build will

• Expose a JSON-RPC 2.0 HTTP interface for tool-based searches.
• Support Server-Sent Events (SSE) for streaming results.

the MCP server offers a search tool powered by SerpAPI to fetch web results about a given topic. Clients can either make a one-off JSON-RPC call or subscribe to a streaming endpoint for real-time updates.

Before we go forward, register, login and create a serpAPI key. You can do it here.

The following is a high level architecture of what we will be building. In this case the user/client is our RAGOrchestrator.

The following are key dependencies we will need, you will find them in the requirements.txt.

#!/usr/bin/env python3
import os
import json
import logging
from typing import List, Dict, Any
import httpx
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from mcp.types import Tool, TextContent

Let me explain why we need these libraries:

• FastAPI: I chose this because it’s incredibly fast, has automatic API documentation, and handles async operations beautifully. Plus, the type hints make debugging a breeze!
• httpx: This is my go-to for async HTTP requests. Unlike the traditional requests library, httpx is built for async/await patterns from the ground up
• uvicorn: This ASGI server can handle thousands of concurrent connections. Perfect for our streaming endpoints!
• mcp.types: Using the official MCP types ensures our server plays nicely with other MCP-compatible tools

FastAPI Setup

Lets initialize the FastAPI server, to run the server

# Configure logging first - this is crucial for debugging
# I always set this up early because you'll need it when things go wrong!
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create our FastAPI application - this is the heart of our server
app = FastAPI()

# Server configuration - I'm being explicit about these settings
HTTP_HOST = "0.0.0.0"  # Listen on all interfaces (allows external connections)
HTTP_PORT = 8000       # Standard development port (change for production)

# External service configuration - keeping API keys in environment variables
# This is a security best practice - never hardcode secrets!
SERPAPI_KEY = os.environ.get("SERPAPI_KEY")
SERPAPI_URL = "https://serpapi.com/search"

app = FastAPI()
HTTP_HOST = "0.0.0.0"
HTTP_PORT = 8000

@app.on_event("startup")
async def startup_event():
    logger.info("Starting MCP Search Server...")
    host_display = HTTP_HOST if HTTP_HOST != "0.0.0.0" else "localhost"
    print(f"Listening at http://{host_display}:{HTTP_PORT}/", flush=True)

• Logging first: I always configure logging before anything else. Trust me, when your server starts misbehaving at 2 AM, you’ll be grateful for detailed logs!
• Host 0.0.0.0: This allows connections from any IP address. In development, this means you can test from other devices on your network
• Environment variables: I never put API keys directly in code. It’s a security nightmare and makes it impossible to share your code safely
• Startup events are perfect for initialization: They run exactly once when the server starts, making them ideal for setup tasks
• Dual logging approach: I use both logger.info() for structured logs and print() for immediate console feedback
• User-friendly URLs: Showing localhost instead of 0.0.0.0 makes it easier for developers to copy-paste the URL
• flush=True: This ensures the message appears immediately in the console, even in Docker containers

Lets now implement the JSON-RPC Endpoint Handling. The / POST endpoint implements a JSON-RPC 2.0 entrypoint: (You can find the documentations along with the code below, so that you understand hte logic)

@app.post("/")
async def mcp_entrypoint(request: Request):
    """Main MCP JSON-RPC endpoint."""
    # Step 1: Parse the incoming request
    # I always start by safely extracting the request body
    body = await request.body()
    try:
        rpc = json.loads(body)
    except Exception:
        # This is crucial - if someone sends malformed JSON, we need to respond properly
        # JSON-RPC has specific error codes, and -32700 means "Parse error"
        response = {"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}}
        logger.info(f"Response: {json.dumps(response, indent=2)}")
        return response

    # Step 2: Extract method and prepare response structure
    # Every JSON-RPC request has a 'method' field - this tells us what to do
    method = rpc.get("method", "")
    # I always prepare the response structure early with required fields
    # The 'id' field links our response back to the original request
    response: Dict[str, Any] = {"jsonrpc": "2.0", "id": rpc.get("id")}

    # Step 3: Route to appropriate handler
    # This is where I implement the method dispatch pattern
    if method == "initialize":
        # The initialize method is like a handshake - clients use it to connect
        response["result"] = {"server": "search-server"}
        logger.info("Initialization received.")
        
    elif method == "list_tools":
        # This is how AI models discover what our server can do
        # I'll show you the list_tools function later - it's pretty cool!
        response["result"] = await list_tools()
        logger.info("Tool list served.")
        
    elif method == "search":
        # This is the main event - where we actually perform searches!
        params = rpc.get("params", {})
        # I'm flexible with parameter names - accept both 'query' and 'topic'
        topic = params.get("query", params.get("topic", ""))
        max_results = params.get("max_results", 5)  # Default to 5 results
        
        if not topic:
            # Proper error handling - JSON-RPC code -32602 means "Invalid params"
            response["error"] = {"code": -32602, "message": "Missing 'query' or 'topic' parameter."}
            logger.warning("Search call with missing topic/query parameter.")
        else:
            logger.info(f"Received search for topic: {topic}")
            # Here's where the real work happens - I'll show you this function next!
            result = await search_for_topic(topic, max_results)
            response["result"] = result
            
    elif method == "shutdown":
        response["result"] = "Server shutdown initiated (not actually shutting down in this demo)."
        logger.info("Shutdown requested (ignored in HTTP version).")
        
    else:
        response["error"] = {"code": -32601, "message": f"Method '{method}' not found."}
        logger.warning(f"Unknown method '{method}' requested by client.")

    # Step 4: Log and return response
    logger.info(f"Response: {json.dumps(response, indent=2)}")
    return response

• Method dispatch based on the method field.
• Returns either a result or a JSON-RPC error.
• Error Handling: Proper JSON-RPC error codes (-32700 for parse errors, -32601 for method not found)
• Logging: Comprehensive logging for debugging and monitoring
• Parameter Flexibility: Accepts both query and topic parameters
• Response Structure: Always includes jsonrpc version and request id

The following table lists the various errors, for your reference:

Creating Server-Sent Events Streaming

Now for the exciting part — real-time streaming! SSE allows us to push search results to clients as they become available.

SSE Endpoint Setup

@app.get("/stream")
async def mcp_stream(topic: str, max_results: int = 5):
    """
    SSE endpoint for streaming search results.
    Client connects to /stream?topic=...&max_results=...
    """
    generator = search_for_topic_stream(topic, max_results)
    return StreamingResponse(generator, media_type="text/event-stream")

The Streaming Generator

Here’s where the actual logic is implemented for SSE streaming :

async def search_for_topic_stream(topic: str, max_results: int = 5):
    """
    Async generator to stream search results as SSE events.
    """
    # Step 1: Validate API key
    if not SERPAPI_KEY or SERPAPI_KEY == "YOUR_SERPAPI_KEY":
        yield f"data: No SerpAPI key found. Set the SERPAPI_KEY environment variable.\n\n"
        return
    
    try:
        # Step 2: Make the search request
        async with httpx.AsyncClient() as client:
            params = {
                "api_key": SERPAPI_KEY,
                "engine": "google",
                "q": topic,
                "num": max_results
            }
            response = await client.get(SERPAPI_URL, params=params)
            response.raise_for_status()
            data = response.json()
            
            # Step 3: Process and stream results
            organic_results = data.get("organic_results", [])
            if not organic_results:
                yield f"data: No specific information found for '{topic}'. Try different keywords.\n\n"
            else:
                # Stream each result individually
                for i, item in enumerate(organic_results[:max_results]):
                    event_data = {
                        "index": i+1,
                        "title": item.get("title"),
                        "link": item.get("link"),
                        "snippet": item.get("snippet", "")
                    }
                    # SSE format: "data: {json}\n\n"
                    yield f"data: {json.dumps(event_data)}\n\n"
                    
    except httpx.HTTPError as e:
        logger.error(f"HTTP error occurred: {e}")
        yield f"data: Failed to search for '{topic}' due to network error\n\n"
    except Exception as e:
        logger.error(f"Unexpected error during streaming: {e}")
        yield f"data: An unexpected error occurred while streaming results for '{topic}'\n\n"

Server-Sent Events have a specific format, as shown below :

data: {"index": 1, "title": "Example Result"}\n\n

• Each event starts with data:
• JSON payload follows
• Each event ends with \n\n (double newline)
• Clients parse this automatically

Tool Discovery System

The LLM models need to know what tools are available. We have to pass the available tools to the LLM, so that they can call back those tools when required. This is a typical implementation of “Tool calling”. Please note that only LLMs that support tool calling can give this kind of callback response. To let the LLM know what tools are available, we have to pass the list of tools that are available in a specific format as shown below.

async def list_tools() -> List[dict]:
    return [
        {
            "name": "search_topic",
            "description": "Search for information about a specific topic",
            "input_schema": {
                "type": "object",
                "properties": {
                    "topic": {"type": "string", "description": "The topic to search for"},
                    "max_results": {"type": "integer", "description": "Maximum number of results to return", "default": 5}
                },
                "required": ["topic"]
            }
        }
    ]

• name: Unique identifier for the tool
• description: Human-readable explanation
• input_schema: JSON Schema defining expected parameters
• required: List of mandatory parameters

This follows the OpenAPI/JSON Schema standard, making it compatible with various AI frameworks.

Search Implementation

Let’s dive into the core search functionality:

async def search_for_topic(topic: str, max_results: int = 5) -> str:
    """
    Search for information about a topic using SerpAPI.
    """
    # Step 1: Validate API key
    if not SERPAPI_KEY or SERPAPI_KEY == "YOUR_SERPAPI_KEY":
        return "No SerpAPI key found. Set the SERPAPI_KEY environment variable."
    
    try:
        # Step 2: Make HTTP request with proper parameters
        async with httpx.AsyncClient() as client:
            params = {
                "api_key": SERPAPI_KEY,
                "engine": "google",        # Use Google search
                "q": topic,               # Search query
                "num": max_results        # Number of results
            }
            response = await client.get(SERPAPI_URL, params=params)
            response.raise_for_status()  # Raise exception for HTTP errors
            data = response.json()
            
            # Step 3: Extract and format results
            organic_results = data.get("organic_results", [])
            if not organic_results:
                return f"No specific information found for '{topic}'. Try searching with different keywords."
            
            # Step 4: Format results for display
            results = []
            for i, item in enumerate(organic_results[:max_results]):
                results.append(f"{i+1}. {item.get('title')}\n{item.get('link')}\n{item.get('snippet', '')}\n")
            
            return "\n".join(results)
            
    except httpx.HTTPError as e:
        logger.error(f"HTTP error occurred: {e}")
        return f"Failed to search for '{topic}' due to network error"
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return f"An unexpected error occurred while searching for '{topic}'"

Here’s the typical response from SerpAPI :

{
  "organic_results": [
    {
      "title": "Example Title",
      "link": "https://example.com",
      "snippet": "This is a description...",
      "displayed_link": "example.com"
    }
  ],
  "search_metadata": {
    "status": "Success",
    "total_time_taken": 1.23
  }
}

We now have the logic to search the web for the given topic, and generate a JSON response of the result, which is passed back to the caller (in our case RAGOrchestrator)

Now that we have the MCP server working, and supporting both JSON-RPC and SSE.

Lets now modify our RAGOrchetrator. If you not read the Part 2, now is the right time, to go through that, where I have walked through the Ranking RAG implementation, and RAGOrchestrator is the key object that implements the agentic RAG logic. Please go through that, before you proceed. You can find it here.

RAGOrchestrator — MCP integration

The RAGOrchestrator is an intelligent information retrieval system that combines:

• Traditional document-based RAG (Retrieval-Augmented Generation)
• Real-time web search via MCP (Model Context Protocol) servers
• LLM-driven decision making using OpenAI’s function calling, that also reranks.

The key innovation is that the LLM itself decides which information source to use based on the question context, rather than following rigid programmatic rules.

The following are the agents that we implemented in Part 2.

PDFLoaderAgent → Document chunking and preprocessing
EmbeddingAgent → Vector embeddings and FAISS indexing
RetrievalAgent → Similarity search with diversity
QAAgent → Answer generation from context
RankingAgent → Multi-candidate answer evaluation
WebSearchAgent → MCP server communication
RAGOrchestrator → Central coordination and decision making

The following diagram show how our agents will interact and integrate with the MCP server

the following interaction diagram, shows the design. Its pretty straight forward. For this example we will be using JSON-RPC

The following is the detailed logic, that we will be implementing.

Before we query, we need to define the tools, this is very critical. This is where things get really exciting! Instead of hardcoding when to use web search vs documents, we will let LLM function calling decide, using Tool Calling (Please note, we can only use LLMs that support tool calling)

def query(self, question: str) -> str:
    print(f"[RAGOrchestrator] Querying for question: {question}")
    
    # Step 1: Define available tools for the LLM
    # This is like giving the LLM a toolbox and letting it choose!
    tools = [
        {
            "type": "function",
            "function": {
                "name": "search_web",
                "description": "Search the web for current information when the question requires recent data or information not in documents",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Search query for web search"
                        }
                    },
                    "required": ["query"]
                }
            }
        }
    ]

• Descriptive tool definitions: The LLM understands when to use each tool based on clear descriptions
• JSON Schema validation: Parameters are properly typed and validated
• Flexible tool selection: The LLM can choose the most appropriate tool for each question

Here’s how I set up the conversation with the LLM:

# Step 2: Set up the conversation with tool calling capability
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant that can search the web for current information or search through documents. Use the appropriate tool based on the question."
        },
        {
            "role": "user",
            "content": question
        }
    ]
    
    try:
        # Step 3: Let the LLM decide what to do!
        response = openai.chat.completions.create(
            model=CHAT_MODEL,                    # Using GPT-3.5-turbo
            messages=messages,
            tools=tools,                         # Available tools
            tool_choice="auto",                  # Let LLM decide
            temperature=0.2,                     # Low randomness for consistency
            max_tokens=500
        )
        
        message = response.choices[0].message
        
        # Step 4: Check what the LLM decided to do
        if message.tool_calls:
            # LLM wants to use a tool - let's handle it!
            return self._handle_tool_calls(message, messages, tools, question)
        else:
            # LLM can answer directly from its knowledge
            return message.content.strip()
            
    except Exception as e:
        print(f"[RAGOrchestrator] Error in tool calling: {e}")
        # If something goes wrong, fall back to document search
        return self._fallback_query(question)

• Intelligent routing: The LLM analyzes the question and chooses the best information source
• Graceful fallbacks: If tool calling fails, we have a backup plan
• Conversational context: The system maintains context throughout the interaction

MCP Integration: WebSearchAgent

Now let me show you how WebSearchAgent communicates with MCP servers. This is where we get real-time web search capabilities:

class WebSearchAgent:
    """Agent to query a MCP server for latest web results related to a question."""
    def __init__(self, endpoint="http://localhost:8000/", apikey=None):
        # I store the MCP server endpoint and API key
        self.endpoint = endpoint
        self.apikey = apikey
    def search_web(self, query: str) -> str:
        print(f"[WebSearchAgent] Querying MCP server for web results: {query}")
        try:
            # Step 1: Build the JSON-RPC 2.0 request
            # This follows the MCP protocol specification
            search_request = {
                "jsonrpc": "2.0",                    # Protocol version
                "id": 2,                             # Request ID for tracking
                "method": "search",                  # The method we want to call
                "params": {
                    "query": query                   # Our search query
                }
            }
            
            # Step 2: Send the request to the MCP server
            resp = requests.post(self.endpoint, json=search_request)
            print("[WebSearchAgent] SEARCH RESPONSE:", resp.json())
            
            # Step 3: Handle HTTP errors
            resp.raise_for_status()
            data = resp.json()
            
            # Step 4: Extract results from the response
            # MCP servers might return results in different formats
            results = data.get('results') or data.get('answer') or str(data)
            
            print("[WebSearchAgent] Web result from MCP server:")
            print(results)
            return results
            
        except Exception as e:
            print(f"[WebSearchAgent] MCP server error: {e}")
            return "[WebSearchAgent] Failed to obtain web data."

• JSON-RPC 2.0 compliance: Following the standard ensures compatibility with any MCP server
• Flexible response parsing: Different MCP servers might structure responses differently
• Comprehensive error handling: Network issues shouldn’t crash the entire system
• Detailed logging: Essential for debugging MCP server interactions

the following parses the MCP Server response and cleanup, and extract what is critical for us to be used as part of the RAG:

def _process_mcp_response(self, mcp_response: str) -> str:
    """Process MCP server JSON response and extract relevant content."""
    try:
        # Step 1: Try to parse as JSON if it's a string
        if isinstance(mcp_response, str):
            try:
                data = json.loads(mcp_response)
            except json.JSONDecodeError:
                # If it's not JSON, return as-is
                return mcp_response
        else:
            data = mcp_response
        
        # Step 2: Extract content from common response fields
        if isinstance(data, dict):
            content = (data.get('result') or 
                      data.get('results') or 
                      data.get('answer') or 
                      data.get('content') or
                      data.get('text'))
            if content:
                return str(content)
                
        return str(data)
    except Exception as e:
        print(f"[_process_mcp_response] Error: {e}")
        return str(mcp_response)

• Format flexibility: Different MCP servers return data in different structures
• Robust parsing: Handles both JSON and plain text responses
• Fallback safety: Always returns something, even if parsing fails

Tool Call Handler: The Decision Engine

When the LLM needs help from any tools, it responds back with a message asking the agent to call the specific tool. The following code, is where we are checking if the LLM wants us to call a tool, and we then call the specific tool.

            response = openai.chat.completions.create(
                model=CHAT_MODEL,
                messages=messages,
                tools=tools,
                tool_choice="auto",
                temperature=0.2,
                max_tokens=500
            )
            
            message = response.choices[0].message
            
            # Check if LLM wants to use tools
            if message.tool_calls:
                return self._handle_tool_calls(message, messages, tools, question)
            else:
                return message.content.strip()

_handle_tool_calls() is the key method where everything comes together! The tool call handler processes the LLM’s decisions and executes the appropriate actions:

def _handle_tool_calls(self, message, messages, tools, question):
    """Handle tool calls from LLM."""
    # Add the LLM's message (with tool calls) to our conversation
    messages.append(message)
    print(f"[RAGOrchestrator] Handling the tool calling: {message}")
    
    # Process each tool call the LLM requested
    for tool_call in message.tool_calls:
        function_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        
        if function_name == "search_web":
            print(f"[RAGOrchestrator] Calling web search tool")
            # Execute web search via MCP server
            result = self.web_agent.search_web(arguments["query"])
            print(f"[RAGOrchestrator] Got web search response: {result}")
            tool_context = self._process_mcp_response(result)
            print(f"[RAGOrchestrator] Created web context for LLM: {tool_context}")
            
        elif function_name == "search_documents":
            print(f"[RAGOrchestrator] Calling document search with ranking")
            if not self.retriever:
                tool_context = "No documents have been ingested."
            else:
                # Execute our sophisticated document search with ranking
                tool_context = self._search_documents_with_ranking(arguments["query"])
            print(f"[RAGOrchestrator] Created document context for LLM: {tool_context}")
            
        else:
            tool_context = "Unknown function called."
        
        # Add the tool result back to our conversation
        # This follows OpenAI's function calling protocol
        messages.append({
            "tool_call_id": tool_call.id,    # Links back to the original tool call
            "role": "tool",                   # Indicates this is a tool response
            "name": function_name,            # Which tool provided this result
            "content": tool_context           # The actual result
        })
    
    print(f"[RAGOrchestrator] Calling LLM back with complete context")
    
    # Now get the final answer from the LLM using all the tool results
    final_response = openai.chat.completions.create(
        model=CHAT_MODEL,
        messages=messages,        # Includes original question + tool results
        temperature=0.2,
        max_tokens=500
    )
    
    return final_response.choices[0].message.content.strip()

• Conversational flow: Tool results are fed back into the same conversation
• Multiple tool support: Can handle multiple tool calls in a single query
• Protocol compliance: Follows OpenAI’s function calling specification exactly
• Context preservation: The LLM sees both the original question and tool results

Result

Find below the video of the screencast, that I recorded. I am uploading my profile, and asking 3 questions.

• First query: “Who is A B Vijay Kumar” : This can be answered with the context that is there in the document (RAG)
• Second query: “Get the latest information about A B Vijay Kumar”: This query requires internet, as I am asking for latest information. you can see how it is able to fetch from the MCP server, and respond with my recent linked in posts.
• Third query: “Who is a IBM Fellow”: is a general knowledge, that LLM can answer by itself.

Wrap up

Finally, we have 3 parts coming together. in Part 1, we went through the concepts of Agentic RAG, and how Agentic RAG is a better approach than the traditional approach. In Part 2, we built a ReRanking RAG Agentic AI solution, and in Part 3, we powered it up by integrating with MCP server. We can extend this further. Please note that most of the code, I wrote are using basic Python libraries. It is much easier to build this system if we use Crew.AI or PydanticAI. They provide very simple wrappers and easy SDK. May be this is a topic for the blog in future :-D

In the meantime, download the code and play around, please leave your feedback and observations.

I am already working on a detailed blog series on MCP, which is coming next…until then bye bye and stay safe.

References

• My Github repo (https://github.com/abvijaykumar/agentic-rag-mcp)
• Screencast (https://youtu.be/NXtBE9DmwCw)