如何在 langchain 中使用 MCP？

实现完整 MCP 三大核心特性：

@mcp.tool：RAG 知识库检索工具；
@mcp.resource：读取配置文件、预设静态上下文；
@mcp.prompt：参数化系统提示词模板；

传输模式：**stdio（子进程管道，无需暴露端口，不用鉴权）**；
Client：langchain‑mcp‑adapters对接 MCP，搭配 LangGraph‑React‑Agent；
项目结构：

rag_mcp_project/
    mcp_server/
        __init__.py          # 空包文件，启用相对导入
        embedding.py
        vector_store.py
        server.py            # MCP‑Server：tool + resource + prompt
    agent/
        __init__.py
        main.py              # LangChain‑MCP‑Client 入口

安装依赖

pip install langchain langchain-openai langgraph fast-mcp langchain-mcp-adapters sentence-transformers chromadb python-dotenv

1. mcp_server/embedding.py

from langchain_community.embeddings import HuggingFaceEmbeddings

embedding = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    model_kwargs={"device": "cpu"},
    encode_kwargs={"normalize_embeddings": True}
)

2. mcp_server/vector_store.py

from langchain_chroma import Chroma
from .embedding import embedding

knowledge_texts = [
    "1.退货申请：用户提交申请后1‑3个工作日审核完成。",
    "2.保修政策：整机质保1年，人为损坏不在保修范围内。",
    "3.开发票：确认收货后7天内可以申请开具增值税普通发票。"
]

db = Chroma.from_texts(knowledge_texts, embedding, persist_directory="./chroma_db")

def search_knowledge(query: str, top_k: int = 3):
    docs = db.similarity_search(query, k=top_k)
    results = [{"content": doc.page_content, "source": "产品售后知识库"} for doc in docs]
    return results

3. mcp_server/server.py（MCP‑Server 核心，包含 tool、resource、prompt）

from fastmcp import FastMCP
from vector_store import search_knowledge

mcp = FastMCP("after‑sales‑mcp‑server")

# ========== 功能1：注册Tool 给Agent调用 ==========
@mcp.tool(description="查询售后知识库，回答客户问题，返回结构化文档列表")
def rag_search(query: str):
    """
    Args:
        query: 用户的问题
    Returns:
        list: 参考文档数组
    """
    return search_knowledge(query)

# ========== 功能2：Resource：只读资源，读取全局配置信息 ==========
@mcp.resource("config://service_intro")
def get_service_intro() -> str:
    """全局客服介绍，作为固定上下文提供给LLM"""
    return """你是平台售后客服，说话简洁友好，严格依据参考文档回答；文档没有提到的内容不能编造。"""

# ========== 功能3：MCP Prompt模板，统一管理发给LLM的系统提示词 ==========
@mcp.prompt()
def customer_answer_template(user_query: str, knowledge_data: str):
    """
    构造LLM完整提示词，由MCP‑Server统一维护模板，上层Agent无需写prompt字符串
    :param user_query: 用户问题
    :param knowledge_data: 检索到的知识库内容
    """
    system_part = mcp.read_resource("config://service_intro")
    prompt = f"""
{system_part}
参考知识库内容：
{knowledge_data}
用户问题：{user_query}
回答格式：简短分点回复，不要多余空话。
"""
    return prompt

if __name__ == "__main__":
    # stdio模式：作为子进程运行，没有端口，不需要鉴权
    mcp.run(transport="stdio")

4. agent/main.py（LangChain‑MCP‑Client）

逻辑：

连接 MCP 服务，自动获取所有 tools；
Client 主动调用 MCP 的rag_search拿到知识库；
Client 调用 MCP‑Prompt 模板生成最终 prompt 送入 LLM；
基于 LangGraph‑React Agent 自动决策什么时候调用工具。

import asyncio
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

async def main():
# 连接MCP‑Server，自动拉起子进程
client = MultiServerMCPClient({
"rag_service": {
"command": "python",
"args": ["mcp_server/server.py"],
"transport": "stdio"
}
})

async with client:
# 获取MCP暴露的工具，自动转为LangChain标准Tool
tools = await client.get_tools()
llm = ChatOpenAI(
model="gpt-4o-mini",
api_key="your-api-key",
base_url="https://xxx/v1"
)
agent = create_react_agent(llm, tools)

user_question = "怎么申请退货？保修多久？"
# 第一步：Agent调用MCP‑Tool拿到知识库结果
tool_result = await client.call_tool("rag_search", {"query": user_question})
knowledge_str = str(tool_result.content)

# 第二步：调用MCP里注册的Prompt模板获取完整指令（重点演示MCP‑Prompt）
prompt_content = await client.get_prompt(
    "customer_answer_template",
    arguments={"user_query": user_question, "knowledge_data": knowledge_str}
)
system_prompt = prompt_content.messages[0].content

# 第三步：把MCP生成的prompt交给Agent执行对话
agent_response = await agent.ainvoke({
    "messages": [
        ("system", system_prompt),
        ("human", user_question)
    ]
})
for msg in agent_response["messages"]:
    if msg.content:
        print(f"\n{msg.content}")


if __name__ == "__main__":
asyncio.run(main())

5. 运行


python agent/main.py

6. 架构职责划分（重点）

MCP‑Server（下层）负责：

@mcp.tool：工具定义、入参校验、执行 RAG 检索；
@mcp.resource：全局静态配置、知识库元信息；
@mcp.prompt()：统一维护提示词模板；后续修改回答话术，只修改 server，Agent 不用改动；
向量模型只加载一次；Agent 重启不会重复加载模型。

LangChain‑Client（上层）负责：

发现 MCP 提供的工具、资源、Prompt；
调用 LLM、管理对话历史；
由 LangGraph 智能判断何时调用工具。

7. 后续切换 SSE（跨机器部署版，生产模式）

修改 server.py


if __name__ == "__main__":
mcp.run(transport="sse", host="0.0.0.0", port=8001)

启动服务：python mcp_server/server.py

Client 修改连接方式 + 增加 Header 鉴权


client = MultiServerMCPClient({
"rag_service": {
"url": "http://127.0.0.1:8001/sse",
"transport": "sse",
"headers": {"X‑MCP‑Api‑Key": "sk‑mcp‑123456"}
}
})

然后在fast‑mcp的 starlette 中间件做 API‑Key 鉴权。