Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG
Part 6 of agentic AI with LangChain: agentic RAG powered by MCP, where retrieval tools come from an MCP server and LangGraph orchestrates the agent flow.
This article is published in English.
Sources: Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG with LangGraph
This walkthrough rebuilds the path from raw materials to a working system for: Agentic AI with LangChain — Part 6: MCP Powered Agentic RAG with LangGraph. The focus is operable steps, explicit checks, and code that you can drop into a repo without guessing intent. For Overview, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
Technical Concepts
When working through Technical Concepts, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
1. System Configurations
When working through 1. System Configurations, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>
class BaseConfig(BaseSettings):
OPENAI_API_KEY: Optional[str]
PINECONE_API_KEY: Optional[str]
TAVILY_API_KEY: Optional[str]
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
2. Project Structure
When working through 2. Project Structure, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
project/
├── rag_process/ # Pinecone RAG operations
│ ├── __init__.py
│ ├── extractor.py
| └── service.py
│
├── graphs # implement ReAct and Reflection StateGraph
│ ├── nodes.py # graph nodes. Defined as Python functions
│ ├── graph.py # graph workflow for MCP in SES mode
│ ├── graph_stdio.py # graph workflow for MCP in STDIO mode
│ ├── consts.py # define node names as constants
│ └── state.py # typed dictionary to store graph states
│
├── mcp_client_sse.py # mcp client in sse mode
├── mcp_client_stdio.py # mcp client in stdio mode
├── server_sse.py # mcp server in sse mode
│── server_stdio.py # mcp server in stdio mode
|── schemas.py # pydantic data models
|── upload.py # python functions to upload files asynchronously
├── main.py # fastAPI app
├── config.py # pydantic BaseConfig
│── utils.py # utility extracting question/docs from message list
├── .env # environment variable definition
├── client.py # streamlit client
├── .gitignore
└── requirements.txt # package requirements
3. MCP (Model Context Protocol)
When working through 3. MCP (Model Context Protocol), write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Cache stable system instructions and tool schemas. Re-sending identical preamble is a common source of burn.
3.1. Concept
When working through 3.1. Concept, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
3.2 MCP Architecture
When working through 3.2 MCP Architecture, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
3.3. MCP Transport Modes
When working through 3.3. MCP Transport Modes, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
3.3.1. STDIO Mode
When working through 3.3.1. STDIO Mode, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface. When working through 3.3.1. STDIO Mode, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
3.3.2. SSE Mode
3.3.2. SSE Mode works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
4. Implementing MCP in STDIO Mode
- Implementing MCP in STDIO Mode works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
4.1. MCP Server
4.1. MCP Server works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move. 4.1. MCP Server works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
# initialize an STDIO FastMCP instance
mcp = FastMCP("Search Server")
# create mcp tools
@mcp.tool()
async def search_agentic_docs(query: str) -> str:
"""
Good for questions about generative agents, prompt engineering, and adversarial attacks.
Pass a natural language search query to retrieve context from the agentic ai database.
"""
# Simply invoke your existing vector service retriever
docs = await vector_service.retriever.ainvoke(query)
# Flatten the document content into a single string for the LLM
return "\n\n".join([d.page_content for d in docs])
# start MCP server in stdio mode
if __name__ == "__main__":
mcp.run(transport="stdio")
4.2. MCP Client
For 4.2. MCP Client, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
# define location of server file
BASE_DIR = Path(__file__).parent
SERVER_SCRIPT = BASE_DIR / "server.py"
# Local server running directly via Python over stdio
SERVER_CONFIG = {
"local_search_server": {
"transport": "stdio",
"command": "python",
"args": [str(SERVER_SCRIPT)],
},
}
# create a MCP client and return
def get_mcp_client() :
"""Works for stdio, sse, or a mix of both!"""
client = MultiServerMCPClient(SERVER_CONFIG)
return client
4.3. Using MCP Tools in the Agent
For 4.3. Using MCP Tools in the Agent, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
from mcp_client_stdio import get_mcp_client
async def build_graph():
"""
build StateGraph
:return:
"""
# initialize a MCP client instance
client = get_mcp_client()
# retrieve MCP tools
mcp_tools = await client.get_tools()
# create an llM
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools
# bind tools to ToolNode and llm
tool_node = ToolNode(tools)
workflow = StateGraph(GraphState)
llm_with_tools = llm.bind_tools(tools)
4.4. Testing the STDIO Workflow
For 4.4. Testing the STDIO Workflow, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap. For 4.4. Testing the STDIO Workflow, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
question_1 = "what is prompt engineer ?"
async def main():
print("hello ReAct agent by Langraph")
message = HumanMessage(content=question_1)
app = await build_graph()
res = await app.ainvoke({"messages": [message]})
if __name__ == "__main__":
asyncio.run(main())
5. Implementing MCP in SSE Mode
When working through 5. Implementing MCP in SSE Mode, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
5.1. MCP Server
When working through 5.1. MCP Server, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
# for SSE
mcp = FastMCP(
name="My MCP Server",
host="0.0.0.0",
port=8001
)
# implement mcp tools here
if __name__ == "__main__":
mcp.run(transport="sse")
5.2. MCP Client
When working through 5.2. MCP Client, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface. When working through 5.2. MCP Client, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
# specify config of the SSE MCP server that the client will connect to
SERVER_CONFIG = {
# config for mcp by sse transportation
"fastapi_mcp_server": {
"transport": "sse",
"url": "http://localhost:8001/sse",
}
}
# create a client
def get_mcp_client() :
client = MultiServerMCPClient(SERVER_CONFIG)
return client
5.3. Using MCP Tools in StateGraph
5.3. Using MCP Tools in StateGraph works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
async def build_graph(mcp_tools: list):
"""
build StateGraph
:return:
"""
# create an llm
llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
tools = [TavilySearch(max_results=3, tavily_api_key=TAVILY_API_KEY)] + mcp_tools
# bind tools to ToolNode and llm
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)
5.4. Why the SSE Implementation Differs from STDIO
5.4. Why the SSE Implementation Differs from STDIO works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
6. Integrating SSE MCP Client with FastAPI
from mcp_client_sse import get_mcp_client
# CREATES a manager for FastAPI startup/shutdown
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):
# create an MCP client, and retrieve MCP tools
mcp_client = get_mcp_client()
mcp_tools = await mcp_client.get_tools()
# Compile graph dynamically with tools and store it on app state
fastapi_app.state.agent = await build_graph(mcp_tools)
# PAUSES here while FastAPI runs and handles requests
yield
app = FastAPI(lifespan=lifespan)
7. Rebuild the StateGraph in Part 5
7.1. FastAPI Endpoint
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(request: Request, body: RAGRequest=Body(...)) -> RAGResponse:
try:
# wrap input query in a HumanMessage
message = HumanMessage(content=body.question)
# loads the StateGraph agent from the request
# this agent is created and maintained by lifespan context
agent = request.app.state.agent
# Async invokes the agent and returns the response
generation = await agent.ainvoke({"messages": [message]})
response = {"answer": generation.get("generation", "")}
return RAGResponse(**response)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))