首发完整版教程，MCP 集成至 LlamaIndex 的技术实践
▌一、前言
本文主要介绍了如何将 MCP（Model Context Protocol，模型上下文协议）工具转换为可以直接使用的 LlamaIndex 工具，使 LlamaIndex 用户能像使用 Claude, Cursor 等现代 AI 应用一样无缝集成这些服务。
▌二、技术背景
2.1 什么是 MCP 协议 ？
MCP（模型上下文协议，https://modelcontextprotocol.io）是面向 AI 应用交互的服务构建协议。开发者可通过该协议构建具备数据资源（resources）、功能工具（tools）和提示模板（prompts）的服务端点。例如，我们可以定义获取 IP 详细信息的 fetch_ipinfo 工具：
@mcp.tool()
async def fetch_ipinfo(ip: str | None = None, **kwargs) -> IPDetails:
"""获取指定IP的详细信息
参数:
ip(str or None): 目标 IP 地址，格式如 192.168.1.1
**kwargs: 传递给 IPInfo 处理器的附加参数
返回:
IPDetails: 包含城市、时区等信息的结构化数据
"""
handler = ipinfo.getHandler(
access_token=os.environ.get("IPINFO_API_TOKEN"),
headers={"user-agent": "basic-mcp-server"},
**kwargs,
)
details = handler.getDetails(ip_address=ip)
return IPDetails(**details.all)
2.2 LlamaIndex 框架特征
LlamaIndex（https://github.com/run-llama/llama_index）是构建知识索引系统的核心框架，其核心价值在于将外部信息结构化地接入大语言模型，实现高效的信息检索、智能问答和对话生成。当需要集成动态数据源时，将 MCP 工具转换为 LlamaIndex 工具成为关键技术需求。
在 LlamaIndex 中，NebulaGraph 已经有两种方法实现 GraphRAG：
KnowledgeGraphIndex用来对任何私有数据从零构建知识图谱（基于 LLM 或者其他语言模型），再 4 行代码进行 Graph RAG：graph_store = NebulaGraphStore( space_name=space_name, edge_types=edge_types, rel_prop_names=rel_prop_names, tags=tags, ) storage_context = StorageContext.from_defaults(graph_store=graph_store) # Build KG kg_index = KnowledgeGraphIndex.from_documents( documents, storage_context=storage_context, max_triplets_per_chunk=10, space_name=space_name, edge_types=edge_types, rel_prop_names=rel_prop_names, tags=tags, ) kg_query_engine = kg_index.as_query_engine()
KnowledgeGraphRAGQueryEngine则可以在任何已经存在的知识图谱上进行 GraphRAG:graph_store = NebulaGraphStore( space_name=space_name, edge_types=edge_types, rel_prop_names=rel_prop_names, tags=tags, ) storage_context = StorageContext.from_defaults(graph_store=graph_store) graph_rag_query_engine = KnowledgeGraphRAGQueryEngine( storage_context=storage_context, )
▌三、为什么需要将 MCP 集成至 LlamaIndex
在实际 AI 应用开发中，集成 MCP 工具主要解决以下需求：
-
功能复用：利用现有 MCP 生态（项目列表见https://github.com/AlexMili/Awesome-MCP）快速扩展系统能力
-
流程统一：建立标准化工具调用范式，简化大模型对外部 API 的调用复杂度
-
自动化管理：通过自动转换机制实现 MCP 工具到
LlamaIndex FunctionTool对象的映射，便于 Agent 自动调度
该集成方案可以有效降低开发成本，同时发挥 LlamaIndex 在上下文管理和对话生成方面的优势。
▌四、技术实现方案
实现流程包含三个关键阶段：
-
MCP 服务通信：通过 MCPClient 类建立服务连接，支持工具发现（list_tools）和调用执行（call_tool）
-
适配器构建：定义 MCPToolAdapter 类，将 MCP 工具元数据转换为 LlamaIndex 的 FunctionTool 对象
-
智能代理集成：在 LlamaIndex Agent 中加载转换后的工具，实现自动化调用
▌五、核心代码解析
协议转换适配器
from llama_index.core.tools import FunctionTool
from pydantic import create_model, BaseModel
class MCPToolAdapter:
def __init__(self, client):
self.client = client
async def list_tools(self):
tools = await self.client.list_tools()
return [
FunctionTool.from_defaults(
fn=self._create_tool_fn(tool.name),
name=tool.name,
description=tool.description,
fn_schema=self._create_pydantic_model(tool.schema),
) for tool in tools
]
def _create_pydantic_model(self, json_schema):
# 实现JSON Schema到Pydantic模型的动态转换
...
类型系统映射
JSON_TYPE_MAPPING = {
"string": str,
"number": float,
"integer": int,
"boolean": bool,
"object": dict,
"array": list
}
def create_dynamic_model(schema):
fields = {}
for name, prop in schema["properties"].items():
py_type = JSON_TYPE_MAPPING[prop["type"]]
if name in schema.get("required", []):
fields[name] = (py_type, Field(...))
else:
fields[name] = (Optional[py_type], Field(None))
return create_model("DynamicModel", **fields)
服务调用封装
async def fetch_ipinfo_wrapper(**kwargs):
return await mcp_client.call_tool("fetch_ipinfo", kwargs)
关键技术创新点
-
动态模式转换：实现 JSON Schema 到 Pydantic 模型的运行时转换，保证参数校验的严格性
-
异步调用管道：基于 Python asyncio 构建非阻塞式服务调用，提升工具执行效率
-
协议抽象层：通过适配器模式解耦 MCP 协议与 LlamaIndex 框架，保证系统扩展性
将 MCP 工具的输入参数转换为 Pydantic 模型的原因在于：MCP 工具使用 JSON Schema 格式定义输入参数（在工具对象的
inputSchema中，类型为Dict[str, Any]），而create_model_from_json_schema函数将其转换为 Pydantic 模型，该模型用作 LlamaIndex 中 FunctionTool 的fn_schema。
▌六、LlamaIndex 中的 MCP 工具
调用实践
以下代码来自 llamaindex_mcp_example.py，完整展示了 MCPToolAdapter 与 LlamaIndex 的集成方式：
import os
from llama_index.core.llms import ChatMessage
from llama_index.core.agent import ReActAgent, ReActChatFormatter
from llama_index.core.agent.react.prompts import REACT_CHAT_SYSTEM_HEADER
from llama_index.llms.azure_openai import AzureOpenAI
from llamaindex_mcp_adapter import MCPToolAdapter
from mcp_client import MCPClient
import anyio
import dotenv
import argparse
dotenv.load_dotenv()
# 系统级指令模板
SYSTEM_PROMPT = """\
您是一个智能法律助手
在回答用户问题前，必须通过IP信息确认用户所在司法管辖区
操作规则：
- 当涉及地理位置、时区等问题时，自动调用fetch_ipinfo工具
- 法律咨询类问题需结合IP所在国家/地区法律体系
"""
async def get_agent(adapter: MCPToolAdapter):
"""智能代理初始化函数"""
tools = await adapter.list_tools() # 获取MCP工具列表
# 配置Azure OpenAI大语言模型
llm = AzureOpenAI(
model=os.environ.get("AZURE_OPENAI_MODEL", "gpt-4"),
max_tokens=int(os.environ.get("AZURE_OPENAI_MAX_TOKENS", 4096)),
azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
api_version=os.environ.get("AZURE_OPENAI_API_VERSION"),
engine=os.environ.get("AZURE_OPENAI_ENGINE")
)
# 构建ReAct智能代理
agent = ReActAgent.from_tools(
llm=llm,
tools=list(tools),
react_chat_formatter=ReActChatFormatter(
system_header=SYSTEM_PROMPT + "\n" + REACT_CHAT_SYSTEM_HEADER,
),
max_iterations=20, # 限制最大推理步数
verbose=True # 启用详细日志
)
return agent
async def handle_user_message(message_content: str, agent: ReActAgent):
"""消息处理流水线"""
user_message = ChatMessage.from_str(role="user", content=message_content)
response = await agent.achat(message=user_message.content)
print(f"[智能代理响应] {response.response}")
if __name__ == "__main__":
async def main():
# 命令行参数解析
parser = argparse.ArgumentParser()
parser.add_argument("--client_type",
choices=["sse", "stdio"],
default="stdio",
help="连接模式：sse(Server-Sent Events)或stdio(标准输入输出)")
args = parser.parse_args()
# MCP客户端初始化
if args.client_type == "sse":
client = MCPClient("http://0.0.0.0:8000/sse") # SSE长连接模式
else:
client = MCPClient( # 子进程模式
"./venv/bin/python",
["mcp_server.py", "--server_type", "stdio"]
)
# 工具适配器初始化
adapter = MCPToolAdapter(client)
agent = await get_agent(adapter)
# 执行测试用例
await handle_user_message("我的IP所在城市是哪里？", agent)
await handle_user_message("持有大麻在我国是否合法？", agent)
anyio.run(main) # 启动异步运行时
关键实现解析
-
工具获取机制：
-
通过
adapter.list_tools()动态加载 MCP 服务端注册的工具列表 -
自动将每个 MCP 工具转换为 LlamaIndex 的
FunctionTool对象 -
智能代理配置：
-
使用 Azure OpenAI 作为底层大语言模型引擎
-
集成
ReAct推理框架，支持多步工具调用 -
设置 20 步的最大推理深度防止无限循环
-
消息处理流程：
-
将用户输入封装为标准化 ChatMessage
-
通过
agent.achat()触发自动化工具调用链
▌七、运行 Demo
7.1 标准输入输出模式
python llamaindex_mcp_example.py --client_type stdio
执行输出示例
[DEBUG] 检测到ListToolsRequest协议请求
[工具调用] 步骤ID: ede14770-91c9-428b... | 输入: 我的IP所在城市是哪里？
[推理日志] 当前用户语言：中文，需调用工具获取数据
[动作触发] fetch_ipinfo | 参数: {}
[网络请求] 调用MCP工具耗时128ms
[观察结果] {"ip": "183.193.123.192", "city": "上海", "country": "CN"...}
[最终响应] 您的IP所在城市为上海
[法律咨询] 步骤ID: e728f5d0-6c0e-454a... | 输入: 持有大麻是否合法？
[推理日志] 已确认用户所在国家：中国
[知识库检索] 加载中国刑法第357条
[最终响应] 根据中国法律，持有大麻属违法行为，最高可判处有期徒刑
7.2 SSE 服务模式
# 终端1：启动SSE服务端
python mcp_server.py --server_type sse
# 终端2：运行客户端
python llamaindex_mcp_example.py --client_type sse
▌八、技术方案总结
通过以下三步实现深度集成：
协议通信层：
-
基于
MCPClient实现服务发现与工具调用 -
支持 SSE/STDIO 双模连接，适应不同部署场景
类型转换层：
-
动态将 JSON Schema 转换为 Pydantic 模型
-
实现参数自动校验与类型提示
def create_dynamic_model(schema):
# 自动生成带字段描述的Pydantic模型
fields = {
name: (map_type(prop), Field(..., description=prop["description"]))
for name, prop in schema["properties"].items()
}
return create_model("DynamicModel", **fields)
智能代理层：
-
集成 ReAct 推理框架实现自动化工具调度
-
通过系统提示词约束工具使用规范
SYSTEM_PROMPT = """\
法律咨询必须遵循以下规则：
1. 首次响应前必须调用fetch_ipinfo确认用户辖区
2. 引用法律条文需标注具体条款编号
3. 不得对未覆盖地区提供法律建议"""
通过持续扩展 MCP 生态，开发者可快速构建具备专业领域能力的智能代理系统。欢迎访问 MCP 开发者社区获取更多工具资源。
查看源代码：https://github.com/psiace/psiace/tree/main/demo/llamaindex-mcp-adapter
✦
如果你觉得 NebulaGraph能帮到你，或者你只是单纯支持开源精神，可以在 GitHub 上为 NebulaGraph 点个 Star！每一个 Star 都是对我们的支持和鼓励✨
https://github.com/vesoft-inc/nebula
✦
✦
