▌一、前言

本文主要介绍了如何将 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

✦

✦