RAG Pipeline vs MCP: Architecture Trade-offs for Production

I have run both RAG pipelines and MCP servers in production on FastAPI. The short version: a RAG pipeline retrieves context for a model to reason over, while an MCP server exposes executable tools the model can call directly. If you need to answer questions from static documents, build RAG. If you need the model to mutate state or call APIs, use MCP. Most production systems eventually need both.

What is the core architectural difference between RAG pipelines and MCP servers?

A RAG pipeline is a read path. It ingests documents, chunks them, embeds them, stores vectors, and retrieves top-k context at query time. The LLM stays passive. It receives context and generates an answer. An MCP server is a write path. It defines a schema of callable functions (tools) with typed inputs and outputs. The LLM becomes an active agent. It decides which tool to call, passes arguments, and receives structured results.

In my stack, the RAG service is a FastAPI app with a /query endpoint. It handles embedding, vector search (usually pgvector or Qdrant), and prompt assembly. The MCP server is a separate FastAPI app exposing /tools/list and /tools/call endpoints (or the stdio transport for local agents). They scale differently. RAG scales with vector index size and embedding latency. MCP scales with tool execution latency and external API rate limits.

I covered the RAG side in detail in Building a Robust RAG Pipeline Architecture for Production. The MCP side introduces a new failure domain: tool timeouts, schema drift, and authentication delegation.

When to use RAG for knowledge retrieval vs MCP for tool execution

Use RAG when the answer exists in a corpus. Legal contracts, product manuals, internal wikis, codebases. The data is mostly static. Updates happen on a schedule (nightly re-index, or CDC from a database). The user asks “What is the refund policy?” The system retrieves chunks and answers.

Use MCP when the answer requires action. “Cancel order 1234.” “Create a Jira ticket.” “Query the production database for last month’s revenue.” The model must call a function. The function has side effects. It needs auth, idempotency keys, and audit logs.

Don’t force RAG to do tool execution by stuffing API specs into the context window. I tried this early on. The hallucination rate on JSON payloads was unacceptable. The model invented required fields. It guessed enum values. MCP solves this by enforcing a contract. The schema is the truth.

Conversely, don’t use MCP for pure lookup. Exposing a search_docs tool that wraps a vector query adds latency (extra hop) and complexity (schema maintenance) for zero gain. Just call the retriever directly from your orchestrator.

How do you combine RAG and MCP in a single agentic workflow?

This is where the rag pipeline vs mcp distinction blurs in practice. A real agent needs both. The pattern I use: the orchestrator (LangGraph or a custom loop) holds a retrieve tool (backed by RAG) and a set of action tools (backed by MCP servers).

from pydantic import BaseModel, Field
from typing import List, Optional
import httpx

class RetrieveInput(BaseModel):
    query: str = Field(..., description="Natural language search query")
    top_k: int = Field(5, ge=1, le=20)

class RetrieveOutput(BaseModel):
    chunks: List[str]
    sources: List[str]

async def retrieve_tool(input: RetrieveInput) -> RetrieveOutput:
    # Direct call to internal RAG FastAPI service
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.post(
            "http://rag-service:8000/query",
            json={"query": input.query, "top_k": input.top_k}
        )
        resp.raise_for_status()
        data = resp.json()
        return RetrieveOutput(chunks=data["chunks"], sources=data["sources"])

# MCP client for action tools
class MCPClient:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {token}"}

    async def call_tool(self, name: str, arguments: dict) -> dict:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(
                f"{self.base_url}/tools/call",
                headers=self.headers,
                json={"name": name, "arguments": arguments}
            )
            resp.raise_for_status()
            return resp.json()

The agent loop:

1. User asks: “Find the SLA for client Acme and create a calendar reminder for the renewal date.”
2. LLM calls retrieve_tool with query “Acme SLA renewal date”.
3. RAG returns chunks containing the date.
4. LLM calls MCP create_calendar_event tool with extracted date.
5. MCP server validates, calls Google Calendar API, returns event ID.
6. LLM summarizes result to user.

This keeps concerns separated. The RAG service knows nothing about calendars. The MCP server knows nothing about vector search. The orchestrator glues them.

What are the implementation patterns using FastAPI for both approaches?

RAG Service Pattern

FastAPI handles the /query endpoint. Background workers handle ingestion. I use a shared Pydantic model for the request/response contract.

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from contextlib import asynccontextmanager
import asyncpg

class QueryRequest(BaseModel):
    query: str
    top_k: int = 5
    filters: dict = {}

class QueryResponse(BaseModel):
    chunks: list[str]
    sources: list[str]
    scores: list[float]

pool: asyncpg.Pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    pool = await asyncpg.create_pool(dsn="postgresql://user:pass@pgvector:5432/db", min_size=2, max_size=10)
    yield
    await pool.close()

app = FastAPI(lifespan=lifespan)

@app.post("/query", response_model=QueryResponse)
async def query_rag(req: QueryRequest):
    # 1. Embed query (call local embedding service or inline)
    # 2. Vector search in pgvector
    # 3. Fetch metadata
    # 4. Return
    pass

Key production concerns: connection pooling for pgvector, embedding model latency (batch if possible), and caching frequent queries. I wrote about chunking strategies that keep retrieval precise in Practical RAG chunking strategies for reliable RAG pipelines.

MCP Server Pattern

FastAPI exposes the MCP HTTP endpoints. Each tool is a registered function with a JSON Schema.

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
from typing import Callable, Any
import inspect
import json

class ToolCall(BaseModel):
    name: str
    arguments: dict

class ToolResult(BaseModel):
    result: Any
    error: str | None = None

# Registry
TOOLS: dict[str, Callable] = {}
SCHEMAS: dict[str, dict] = {}

def tool(func: Callable) -> Callable:
    TOOLS[func.__name__] = func
    # Generate JSON schema from signature + docstring (simplified)
    sig = inspect.signature(func)
    props = {}
    required = []
    for name, param in sig.parameters.items():
        # Real impl uses pydantic TypeAdapter for proper schema gen
        props[name] = {"type": "string"}
        if param.default == inspect.Parameter.empty:
            required.append(name)
    SCHEMAS[func.__name__] = {
        "name": func.__name__,
        "description": func.__doc__ or "",
        "inputSchema": {"type": "object", "properties": props, "required": required}
    }
    return func

@tool
async def cancel_order(order_id: str, reason: str = "user_requested") -> dict:
    """Cancel an order by ID."""
    # Call internal order service, handle idempotency
    return {"status": "cancelled", "order_id": order_id}

app = FastAPI()

@app.get("/tools/list")
async def list_tools():
    return {"tools": list(SCHEMAS.values())}

@app.post("/tools/call", response_model=ToolResult)
async def call_tool(body: ToolCall, authorization: str = Header(...)):
    # Validate token (see Security section)
    if body.name not in TOOLS:
        raise HTTPException(404, "Tool not found")
    try:
        result = await TOOLS[body.name](**body.arguments)
        return ToolResult(result=result)
    except Exception as e:
        return ToolResult(result=None, error=str(e))

Run this behind a load balancer. Scale horizontally. The stdio transport (for local Claude Desktop style clients) needs a separate entry point but the tool logic is shared.

Security and authentication considerations for each architecture

RAG Security

The risk is data leakage. The vector store contains chunks of sensitive docs. The embedding model might memorize.

• Auth: The /query endpoint must validate a JWT. Extract user_id and roles.
• Authorization: Apply metadata filters at query time. WHERE doc_id IN (SELECT id FROM docs WHERE owner_id = $user_id). Never trust the client to filter.
• PII: Redact before embedding, or use a local embedding model (no API calls to OpenAI/Cohere with sensitive data).
• Audit: Log every query (user, query hash, top-k doc IDs) for compliance.

MCP Security

The risk is unauthorized action. Tools mutate state.

• Auth: The MCP server validates the token. It must know who is calling.
• Authorization: Fine-grained. The cancel_order tool checks if the caller owns the order. Do this inside the tool function, not just at the gateway.
• Token Forwarding: Often the MCP server calls downstream APIs (Stripe, GitHub, Internal RPC). It needs to exchange the user’s token for a service token (OAuth2 On-Behalf-Of flow) or use a service account with impersonation headers.
• Idempotency: Every mutating tool must accept an idempotency_key. The client (orchestrator) generates it. The MCP server stores it (Redis) with a TTL. Prevents double-charge on retry.
• Rate Limiting: Per-user, per-tool. cancel_order is expensive. read_order is cheap. Configure limits per tool.

Decision matrix for choosing the right pattern for your use case

When NOT to use RAG: You need real-time data (stock prices, sensor readings). Use a tool that calls the live API. When NOT to use MCP: You just need semantic search over a PDF corpus. The HTTP overhead and schema maintenance aren’t worth it. When NOT to combine: Your use case is purely chat-over-docs or purely chat-to-API. Don’t add the other component “just in case.” YAGNI applies hard here.

FAQ

Can I use MCP to implement a RAG retriever tool? Yes, you can wrap your RAG /query endpoint as an MCP tool. This lets a generic MCP client (like Claude Desktop) use your knowledge base. It adds network latency and schema overhead. For your own agents, call the RAG service directly.

How do I handle authentication between the orchestrator, RAG, and MCP? Use a shared secret or mTLS for service-to-service calls (orchestrator -> RAG/MCP). Forward the user’s JWT from the frontend to the orchestrator, then to MCP for downstream auth. RAG uses the user ID from the JWT for metadata filtering.

What is the biggest operational difference? RAG breaks silently (bad chunks, stale index, drift). MCP breaks loudly (500 errors, timeouts, failed transactions). Monitor RAG with eval sets (recall@k, faithfulness). Monitor MCP with success rates, latency percentiles, and idempotency key collisions.

Do I need a vector database for MCP? No. MCP is protocol-agnostic. It passes JSON. If your tool needs vector search, it calls a vector DB internally. The MCP server is just the API gateway for the tool.

Key Takeaways

• RAG is for retrieval (read). MCP is for execution (write). They solve different problems.
• Don’t stuff API specs into RAG context. Use MCP for structured tool calling.
• Production agents usually need both: a retrieve tool (RAG-backed) and action tools (MCP-backed).
• Secure RAG at the document level (metadata filters). Secure MCP at the operation level (RBAC/ABAC inside tools).
• Enforce idempotency keys on every mutating MCP tool. You will thank yourself during a retry storm.