MCP 与 LangChain/LlamaIndex 深度整合：构建企业级 AI Pipeline

MCP 与框架：互补而非替代

很多人以为 MCP 协议出现后，LangChain 和 LlamaIndex 就没用了。实际上恰恰相反——MCP 解决的是工具标准化问题，框架解决的是编排和组合问题。

简单来说：

• MCP = 工具之间的 USB-C 接口
• LangChain/LlamaIndex = 主板上的电路设计

两者完美互补。

在 LangChain 中使用 MCP

LangChain 从 2026 年初开始原生支持 MCP 协议的 Tool 加载。

安装与基础用法

# langchain-mcp 包
from langchain_mcp import load_mcp_tools
from langchain.agents import create_react_agent

# 从 MCP Server URL 加载工具
mcp_tools = load_mcp_tools("http://localhost:8000")
# 返回标准的 LangChain Tool 对象列表
# 包含 name, description, args_schema, func 等属性

# 直接用于 Agent
agent = create_react_agent(llm, mcp_tools, prompt)

混合使用：MCP + 原生 LangChain 工具

from langchain.tools import tool
from langchain_mcp import load_mcp_tools

# 本地工具（纯 Python 函数）
@tool
def get_current_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}: 25°C, 晴"

# 远程 MCP 工具（数据库查询）
db_tools = load_mcp_tools("http://mcp-postgres:8000")

# 外部 API MCP 工具（Slack/Salesforce 等）
api_tools = load_mcp_tools("http://mcp-gateway:8000")

# 组合使用
all_tools = [get_current_weather, *db_tools, *api_tools]

# 创建 Agent
agent = create_react_agent(llm, all_tools, prompt)

动态工具发现

MCP 的 ListTools 能力让 Framework 层可以动态发现可用工具：

class DynamicToolManager:
    def __init__(self):
        self.mcp_servers = {}
    
    def register_mcp(self, name: str, url: str):
        # 连接 MCP Server 并获取其所有工具
        tools = load_mcp_tools(url)
        self.mcp_servers[name] = {"url": url, "tools": tools}
        return tools
    
    def get_all_tools(self):
        all_tools = []
        for server_name, info in self.mcp_servers.items():
            for tool in info["tools"]:
                # 给工具加前缀避免名称冲突
                tool.name = f"{server_name}_{tool.name}"
                all_tools.append(tool)
        return all_tools

manager = DynamicToolManager()
manager.register_mcp("postgres", "http://mcp-postgres:8000")
manager.register_mcp("github", "http://mcp-github:8001")
tool_list = manager.get_all_tools()

这在微服务架构中非常有用——每个微服务暴露一个 MCP Server，Agent 动态发现所有可用工具。

在 LlamaIndex 中使用 MCP

LlamaIndex 同样支持直接挂载 MCP 工具作为 QueryEngine 或 Tool。

基础集成

from llama_index.core import VectorStoreIndex
from llama_index.mcp import MCPToolSpec

# 加载 MCP 工具作为数据源
mcp_spec = MCPToolSpec(endpoint="http://mcp-server:8000")
tools = mcp_spec.to_tool_list()

# 创建 QueryEngine
from llama_index.core.query_engine import ToolRetrieverRouterQueryEngine
engine = ToolRetrieverRouterQueryEngine.from_defaults(
    tools=tools,
    retriever=retriever,
)

MCP + RAG 的经典架构

用户提问
    │
    ▼
LlamaIndex 路由层
    │
    ├─ 向量检索 (RAG) → 文档片段
    │
    └─ MCP 工具调用
        ├─ mcp-postgres  → 数据库查询结果
        ├─ mcp-slack     → 聊天记录
        └─ mcp-jira      → 项目信息
    │
    ▼
LLM 综合回答

这种架构的核心优势：RAG 负责静态知识，MCP 负责动态数据。前者给 LLM 提供文档上下文，后者给 LLM 提供实时的系统数据。

示例：数据分析 Agent

from llama_index.core.agent import ReActAgent
from llama_index.mcp import MCPToolSpec

# 注册多个 MCP Server
analytics_tools = MCPToolSpec(endpoint="http://mcp-clickhouse:8000").to_tool_list()
dashboard_tools = MCPToolSpec(endpoint="http://mcp-metabase:8000").to_tool_list()
alert_tools = MCPToolSpec(endpoint="http://mcp-pagerduty:8000").to_tool_list()

agent = ReActAgent.from_tools(
    tools=analytics_tools + dashboard_tools + alert_tools,
    llm=llm,
    verbose=True,
    max_iterations=15,
)

# 用户提问："昨天用户转化率下降的原因是什么？"
# Agent 会：
# 1. 调 mcp-clickhouse 查用户行为数据
# 2. 调 mcp-metabase 看仪表板趋势
# 3. 综合分析给出结论
response = agent.query("昨天用户转化率下降的原因是什么？")

LangGraph 中的 MCP 节点

LangGraph 把 LangChain 的 Chain 变成了有向图。MCP 工具可以作为一个 Graph Node：

from langgraph.graph import StateGraph, END
from langchain_mcp import load_mcp_tools

# 加载 MCP 工具
mcp_tools = load_mcp_tools("http://mcp-internal-apis:8000")
tool_dict = {tool.name: tool for tool in mcp_tools}

# 定义图
workflow = StateGraph(AgentState)

# MCP 调用节点
def call_mcp_tool(state):
    tool_name = state["next_tool"]
    tool_input = state["tool_input"]
    result = tool_dict[tool_name].invoke(tool_input)
    return {"tool_result": result}

workflow.add_node("mcp_executor", call_mcp_tool)

# LLM 推理节点
def llm_reason(state):
    # ... 让 LLM 决定下一步用哪个 MCP 工具
    pass

workflow.add_node("llm", llm_reason)

workflow.add_edge("llm", "mcp_executor")
workflow.add_conditional_edges("mcp_executor", router, {
    "continue": "llm",
    "end": END,
})

app = workflow.compile()

这种方式的优势：同一份 MCP 工具可以被不同的 Graph 路径复用，避免工具注册的重复工作。

最佳实践

1. 工具命名规范

当多个 MCP Server 被加载到同一个框架时，工具名可能冲突。推荐命名格式：

{server_type}_{tool_name}

例如：

• postgres_query
• github_list_prs
• slack_send_message

2. 异常处理

MCP Server 可能不响应或返回错误，框架层需要容错：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def safe_mcp_call(tool, input_data):
    try:
        return tool.invoke(input_data)
    except ConnectionError:
        # 尝试从备用 MCP Server 调用
        return fallback_mcp_call(tool.name, input_data)
    except Exception as e:
        log_error(tool.name, str(e))
        return {"error": str(e), "status": "failed"}

3. 缓存策略

对高频、低变化的 MCP 调用做缓存：

from functools import lru_cache

@lru_cache(maxsize=128)
def cached_mcp_call(tool_name: str, input_hash: str):
    tool = tool_dict[tool_name]
    return tool.invoke(json.loads(input_hash))

缓存粒度：按工具+参数哈希，TTL 根据数据特性设定（分钟级到小时级）。

总结

MCP + LangChain/LlamaIndex 的组合代表了 AI 开发的未来方向：

能力	MCP 提供	框架提供
工具发现	ListTools 动态注册	工具路由与组合
参数校验	InputSchema JSON Schema	Pydantic 验证
执行	CallTool 统一接口	链式/图式编排
扩展	新增 MCP Server = 新增能力	Agent/Graph 逻辑复用

MCP 标准化了「工具怎么连」，框架处理了「工具怎么用」。两者加起来，就是可组合、可扩展、可管理企业级 AI Pipeline。