手把手为 Cursor 构建自定义 MCP 服务器集成 Linkup 深度网页搜索与 LlamaIndex RAG 双工具【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub本篇技术指南以仓库 cursor_linkup_mcp 为例完整讲解如何从零构建一个可被 Cursor 直接调用的自定义 MCP 服务器。该服务器对外暴露两个工具一个基于 Linkup 的实时深度网页搜索工具和一个基于 LlamaIndex 事件驱动 Workflow 的本地文档 RAG 工具。读完本文你将掌握 FastMCP 工具注册、stdio 传输接入 Cursor、Linkup API 参数调优以及用 LlamaIndex Workflow 实现检索—合成全流程 RAG 的完整实战方法。1. 项目要解决什么问题现代 AI 编码工具如 Cursor的能力边界往往取决于其能否触达外部数据实时的网页信息、私有的本地文档。本项目的目标非常聚焦——构建一个自定义 MCPModel Context Protocol服务器把它接入 Cursor让 Cursor 获得两种此前不具备的能力深度网页搜索Web Search通过 Linkup 服务执行实时联网搜索并返回带出处的结构化答案本地文档 RAG对data目录中的私有文档示例为DeepSeek.pdf建立向量索引让 Cursor 能基于本地资料回答DeepSeek 是如何训练的这类领域问题。两者的组合在架构上形成了互补Linkup 补足实时、开放网络信息RAG 补足私有、离线知识这正是 Agent 化编码工具最常用的两种知识接入方式。仓库目录结构如下cursor_linkup_mcp/ ├── README.md # 项目说明与上手指南本文依托的原始文档 ├── pyproject.toml # uv 项目配置与依赖声明 ├── uv.lock # 依赖锁定文件 ├── server.py # MCP 服务器入口注册两个工具并启动 ├── rag.py # LlamaIndex 事件驱动 RAGWorkflow 实现 ├── data/ │ └── DeepSeek.pdf # 示例 RAG 知识库文档围绕 DeepSeek └── assets/ └── thumbnail.png # 教程视频封面图核心实现集中在两个 Python 文件server.py 负责 MCP 协议层与工具注册rag.py 负责 RAG 引擎本体两者通过一次进程启动串联先灌入文档再启动服务器。2. 环境要求与依赖清单读懂 pyproject.toml项目使用 uv[project] name cursor-linkup-mcp version 0.1.0 description Add your description here readme README.md requires-python 3.12 dependencies [ ipykernel6.29.5, linkup-sdk0.2.4, llama-index0.12.25, llama-index-embeddings-huggingface0.5.2, llama-index-llms-ollama0.5.3, mcp[cli]1.5.0, ]各依赖在本文架构中的角色如下依赖用途在代码中的消费方linkup-sdk0.2.4调用 Linkup 网页搜索 APIserver.py 的LinkupClientllama-index0.12.25RAG 核心框架索引、Workflow、文档加载rag.py 全部 RAG 逻辑llama-index-embeddings-huggingface0.5.2HuggingFace 本地 embedding 模型rag.py 默认BAAI/bge-small-en-v1.5llama-index-llms-ollama0.5.3经 Ollama 调用本地 LLM 生成回答rag.py 默认llama3.2mcp[cli]1.5.0官方 MCP Python SDKFastMCP 与 CLI 调试工具server.py 的FastMCPipykernel6.29.5Jupyter/交互内核支持非核心运行路径供调试使用两个前置环境约束值得注意其一requires-python 3.12需要 Python 3.12 及以上其二RAG 分支走的是全本地推理链路Ollama LLM HuggingFace embedding无需 GPU 即可运行但对 Ollama 服务可用性有依赖详见第 6 节。依赖版本由uv.lock精确锁定保证uv sync后复现环境一致。3. 环境配置先做三件事3.1 同步依赖按原始文档的指引在项目目录执行uv sync该命令会根据pyproject.tomluv.lock创建虚拟环境并安装全部依赖。因为代码已提交了uv.lock同步结果具备确定性如需后续运行脚本建议通过uv run前缀执行uv run python server.pyuv 会自动复用uv sync创建的同一虚拟环境。3.2 配置环境变量原始文档明确要求设置两个环境变量LINKUP_API_KEY... OPENAI_API_KEY...LINKUP_API_KEY必填由 server.py 的LinkupClient()读取用于网页搜索鉴权。原始文档提示到 Linkup 官网申请。OPENAI_API_KEYREADME 要求一并配置。需要说明的是从当前仓库源码看server.py 与 rag.py 中并未直接 import openai——RAG 生成端实际走的是本地 Ollama。可以推断该变量更多是原始教程基于 OpenAI 后端方案的通用要求若你完全复刻本仓库的本地链路务必保证至少LINKUP_API_KEY有效同时保留OPENAI_API_KEY以备后续切换云端 LLM 或遵循视频原流程亦不会造成冲突。配置文件加载由 server.py 顶部的load_dotenv()完成因此你也可以在项目根目录放一个.env文件来存放以上密钥。3.3 准备 RAG 数据RAG 工具读取的是启动时传入的目录。仓库内置了示例文档data/DeepSeek.pdf关于 DeepSeek 模型的中文/技术资料无需额外准备即可跑通若要测试自己的私有资料将 PDF或 LlamaIndexSimpleDirectoryReader支持的其他格式放入同一目录即可。4. MCP 服务器入口 server.py 逐行拆解server.py 是整个项目的核心入口逻辑非常紧凑——通过 FastMCP 只做三件事初始化、注册工具、启动。4.1 初始化from dotenv import load_dotenv from linkup import LinkupClient from rag import RAGWorkflow from mcp.server.fastmcp import FastMCP load_dotenv() mcp FastMCP(linkup-server) client LinkupClient() rag_workflow RAGWorkflow()FastMCP(linkup-server)创建命名 MCP 服务器。工具名、描述会以 MCP 协议暴露给 Cursor供其 Agent 决定何时调用。LinkupClient()Linkup SDK 的无参客户端API Key 从环境变量注入。RAGWorkflow()实例化 rag.py 中定义的 LlamaIndex Workflow此时尚未灌入文档索引为None。4.2 工具一Linkup 深度网页搜索mcp.tool() def web_search(query: str) - str: Search the web for the given query. search_response client.search( queryquery, depthstandard, # standard or deep output_typesourcedAnswer, # searchResults or sourcedAnswer or structured structured_output_schemaNone, # must be filled if output_type is structured ) return search_responsemcp.tool()装饰器会把下方函数自动声明为 MCP 工具Cursor 侧即可在对话中调用名为web_search、签名含query参数的工具。核心参数query搜索问题由调用方Cursor 的 Agent根据用户意图自动生成。depthstandard搜索深度枚举可选standard或deep。deep模式会执行更深层级的抓取与推理通常带来更全面结果但耗时更长standard适合日常快速检索。output_typesourcedAnswer输出类型三选一searchResults返回原始检索结果列表sourcedAnswer由 Linkup 基于检索结果综合生成答案并附带出处本项目的默认选择兼顾回答质量与可溯源structured返回结构化数据。structured_output_schemaNone仅当output_typestructured时必须提供 schema源码注释明确提示 must be filled if output_type is structured否则保持None。函数 docstringSearch the web for the given query.会作为工具描述同步进 Cursor 的能力列表帮助模型判断什么时候该调用网页搜索因此写好一句话描述对 Agent 命中率很重要。4.3 工具二本地文档 RAGmcp.tool() async def rag(query: str) - str: Use a simple RAG workflow to answer queries using documents from data directory about Deep Seek response await rag_workflow.query(query) return str(response)rag是异步工具内部直接委托给rag_workflow.query(query)。工具描述明确告知模型它回答的是data目录中关于 DeepSeek的文档问题——这一定位让 Cursor 能自发区分该联网搜调 web_search还是该查本地资料调 rag。4.4 启动入口if __name__ __main__: asyncio.run(rag_workflow.ingest_documents(data)) mcp.run(transportstdio)启动序列值得细看先用asyncio.run(rag_workflow.ingest_documents(data))预灌入文档——读取data目录全部文件并构建向量索引失败则后续 rag 查询会报 No documents have been ingested再以mcp.run(transportstdio)启动服务器。stdio 传输是 MCP 客户端Cursor、Claude Desktop 等以子进程方式拉起服务器的标准通道Cursor 启动该命令后通过标准输入/输出与该进程通信。5. rag.py 源码剖析LlamaIndex 事件驱动 RAG 工作流如果说 server.py 是协议层rag.py 就是引擎层。它不是简单的query_engine一行调用而是用 LlamaIndex 的新一代事件驱动 Workflow将 RAG 拆成可观察、可编排的三个步骤。5.1 事件与工作流骨架nest_asyncio.apply() class RetrieverEvent(Event): Result of running retrieval nodes: list[NodeWithScore] class RAGWorkflow(Workflow): def __init__(self, model_namellama3.2, embedding_modelBAAI/bge-small-en-v1.5): super().__init__() self.llm Ollama(modelmodel_name) self.embed_model HuggingFaceEmbedding(model_nameembedding_model) Settings.llm self.llm Settings.embed_model self.embed_model self.index None关键点RAGWorkflow继承llama_index.core.workflow.Workflow用step装饰的异步方法构成有向图步骤间通过自定义事件如RetrieverEvent传递数据。模型默认值写死在构造参数上LLM 为经 Ollama 调用的llama3.2embedding 为本地加载的BAAI/bge-small-en-v1.5HuggingFace 小模型CPU 即可运行。通过Settings.llm/Settings.embed_model将模型写入 LlamaIndex 全局设置后续索引与合成器会自动继承。nest_asyncio.apply()允许嵌套事件循环。这是本地 Jupyter/混合调度场景的经典坑——当外层已有事件循环如 FastMCP 的 async 运行环境时内层再次asyncio.run会冲突nest_asyncio正是为此打补丁。5.2 三个 stepingest → retrieve → synthesize步骤一灌入文档ingeststep async def ingest(self, ctx: Context, ev: StartEvent) - StopEvent | None: dirname ev.get(dirname) if not dirname: return None documents SimpleDirectoryReader(dirname).load_data() self.index VectorStoreIndex.from_documents(documentsdocuments) return StopEvent(resultself.index)用SimpleDirectoryReader加载目录内文档VectorStoreIndex.from_documents在内存中完成切块 embedding 建索引。索引暂存于self.index注意内存索引进程重启后需重新灌入这正是 server.py 每次启动先 ingest 的原因。步骤二检索retrievestep async def retrieve(self, ctx: Context, ev: StartEvent) - RetrieverEvent | None: query ev.get(query) index ev.get(index) or self.index if not query: return None if index is None: print(Index is empty, load some documents before querying!) return None retriever index.as_retriever(similarity_top_k2) nodes await retriever.aretrieve(query) await ctx.set(query, query) return RetrieverEvent(nodesnodes)index.as_retriever(similarity_top_k2)取 top-2 最相关文本块similarity_top_k直接控制送入 LLM 的上下文量可按知识密度上调。空索引保护未灌文档时打印提示并优雅返回None避免下游空指针。ctx.set(query, query)把用户问题写入步骤共享的Context供后续合成步骤读取。步骤三合成回答synthesizestep async def synthesize(self, ctx: Context, ev: RetrieverEvent) - StopEvent: summarizer CompactAndRefine(streamingTrue, verboseTrue) query await ctx.get(query, defaultNone) response await summarizer.asynthesize(query, nodesev.nodes) return StopEvent(resultresponse)使用 LlamaIndex 的CompactAndRefine合成器流式输出、verbose 打印中间过程CompactAndRefine会先尽量把检索块压缩进单次 prompt超出窗口时再采用逐块精炼refine策略累加答案是长上下文场景下的稳健默认选择。5.3 对外封装方法async def query(self, query_text: str): if self.index is None: raise ValueError(No documents have been ingested. Call ingest_documents first.) result await self.run(queryquery_text, indexself.index) return result async def ingest_documents(self, directory: str): result await self.run(dirnamedirectory) self.index result return resultquery/ingest_documents是给 server.py 与独立测试用的门面方法前者把用户问题以queryquery_text, indexself.index塞进StartEvent启动整条流水线后者把目录路径塞进StartEvent触发灌入。rag.py文件底部还附带了独立演示入口main()可脱离 MCP 直接验证 RAG 链路workflow RAGWorkflow() await workflow.ingest_documents(data) result await workflow.query(How was DeepSeekR1 trained?) async for chunk in result.async_response_gen(): print(chunk, end, flushTrue)6. 在 Cursor 中接入并运行 MCP 服务器6.1 以 stdio 子进程方式注册server.py 以transportstdio启动意味着 MCP 客户端需要把该命令当作子进程拉起。在 Cursor 的 MCP 服务器配置中添加一条本地命令型服务器对应type: stdio启动命令可参考uv run python server.py前提是命令的工作目录指向仓库根目录保证能 importrag模块、能加载data目录、能读取.env。Cursor 在识别到该服务器提供的web_search、rag两个工具后便可在对话中按需调用。补充说明mcp[cli]1.5.0依赖同时带入了 MCP 官方命令行工具如mcp run/mcp dev等可在接入 Cursor 前用其对server.py做协议级冒烟测试属官方 SDK 自带的排查手段本项目依赖声明已为其预留了安装条件。6.2 启动后的典型运行链路服务真正跑起来后一次典型问答会这样流转Cursor 的 Agent 判断用户问题需要实时信息 → 调用web_search服务器内LinkupClient.search(query, depth, output_type)发起联网检索返回带出处的综合答案若用户问题指向本地 DeepSeek 资料 → 调用rag→RAGWorkflow.run走retrieve(similarity_top_k2) → CompactAndRefine 合成流水线基于DeepSeek.pdf作答。7. 运行注意要点与排错清单结合源码实现落地时需重点检查以下事项现象排查方向依据rag返回 No documents have been ingestedserver.py主流程必须在mcp.run前完成 ingest确认data目录非空rag.py日志打印 Index is empty...检索前索引为空灌入失败或被跳过rag.pyRAG 调用报事件循环冲突确认nest_asyncio.apply()在导入即执行模块顶层已调用rag.pyRAG 无响应/报连接错误本地 Ollama 服务需运行且已ollama pull llama3.2模型名可在RAGWorkflow(model_name...)处调整rag.pyweb_search报鉴权错误LINKUP_API_KEY未配置或.env未加载load_dotenv()在模块顶层执行server.py需要结构化输出但报错output_typestructured时必须同时传structured_output_schemaserver.py换目录后 RAG 仍答旧文档索引为进程内存态重启进程重新 ingest使新文档生效server.py8. 小结这套架构的可迁移价值从本仓库可以抽象出一个可复用的 MCP Agent 工具接入范式协议层FastMCP 工具注册 能力层外部 API / 本地 RAG 引擎分层解耦。其中 server.py 展示的是一行装饰器 一个 Cursor 可用工具的最小接入成本rag.py 展示的是把 RAG 显式建模为 ingest / retrieve / synthesize 事件流、从而让每一步可观测可替换的思路。想要扩展其他能力时只需沿用mcp.tool()追加新函数并把新引擎以同样的 Workflow 或 client 对象注入即可——这正是 MCP 生态下为 AI 编码工具持续插拔能力的标准姿势。【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 手把手为 Cursor 构建自定义 MCP 服务器：集成 Linkup 深度网页搜索与 LlamaIndex RAG 双工具
手把手为 Cursor 构建自定义 MCP 服务器集成 Linkup 深度网页搜索与 LlamaIndex RAG 双工具【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub本篇技术指南以仓库 cursor_linkup_mcp 为例完整讲解如何从零构建一个可被 Cursor 直接调用的自定义 MCP 服务器。该服务器对外暴露两个工具一个基于 Linkup 的实时深度网页搜索工具和一个基于 LlamaIndex 事件驱动 Workflow 的本地文档 RAG 工具。读完本文你将掌握 FastMCP 工具注册、stdio 传输接入 Cursor、Linkup API 参数调优以及用 LlamaIndex Workflow 实现检索—合成全流程 RAG 的完整实战方法。1. 项目要解决什么问题现代 AI 编码工具如 Cursor的能力边界往往取决于其能否触达外部数据实时的网页信息、私有的本地文档。本项目的目标非常聚焦——构建一个自定义 MCPModel Context Protocol服务器把它接入 Cursor让 Cursor 获得两种此前不具备的能力深度网页搜索Web Search通过 Linkup 服务执行实时联网搜索并返回带出处的结构化答案本地文档 RAG对data目录中的私有文档示例为DeepSeek.pdf建立向量索引让 Cursor 能基于本地资料回答DeepSeek 是如何训练的这类领域问题。两者的组合在架构上形成了互补Linkup 补足实时、开放网络信息RAG 补足私有、离线知识这正是 Agent 化编码工具最常用的两种知识接入方式。仓库目录结构如下cursor_linkup_mcp/ ├── README.md # 项目说明与上手指南本文依托的原始文档 ├── pyproject.toml # uv 项目配置与依赖声明 ├── uv.lock # 依赖锁定文件 ├── server.py # MCP 服务器入口注册两个工具并启动 ├── rag.py # LlamaIndex 事件驱动 RAGWorkflow 实现 ├── data/ │ └── DeepSeek.pdf # 示例 RAG 知识库文档围绕 DeepSeek └── assets/ └── thumbnail.png # 教程视频封面图核心实现集中在两个 Python 文件server.py 负责 MCP 协议层与工具注册rag.py 负责 RAG 引擎本体两者通过一次进程启动串联先灌入文档再启动服务器。2. 环境要求与依赖清单读懂 pyproject.toml项目使用 uv[project] name cursor-linkup-mcp version 0.1.0 description Add your description here readme README.md requires-python 3.12 dependencies [ ipykernel6.29.5, linkup-sdk0.2.4, llama-index0.12.25, llama-index-embeddings-huggingface0.5.2, llama-index-llms-ollama0.5.3, mcp[cli]1.5.0, ]各依赖在本文架构中的角色如下依赖用途在代码中的消费方linkup-sdk0.2.4调用 Linkup 网页搜索 APIserver.py 的LinkupClientllama-index0.12.25RAG 核心框架索引、Workflow、文档加载rag.py 全部 RAG 逻辑llama-index-embeddings-huggingface0.5.2HuggingFace 本地 embedding 模型rag.py 默认BAAI/bge-small-en-v1.5llama-index-llms-ollama0.5.3经 Ollama 调用本地 LLM 生成回答rag.py 默认llama3.2mcp[cli]1.5.0官方 MCP Python SDKFastMCP 与 CLI 调试工具server.py 的FastMCPipykernel6.29.5Jupyter/交互内核支持非核心运行路径供调试使用两个前置环境约束值得注意其一requires-python 3.12需要 Python 3.12 及以上其二RAG 分支走的是全本地推理链路Ollama LLM HuggingFace embedding无需 GPU 即可运行但对 Ollama 服务可用性有依赖详见第 6 节。依赖版本由uv.lock精确锁定保证uv sync后复现环境一致。3. 环境配置先做三件事3.1 同步依赖按原始文档的指引在项目目录执行uv sync该命令会根据pyproject.tomluv.lock创建虚拟环境并安装全部依赖。因为代码已提交了uv.lock同步结果具备确定性如需后续运行脚本建议通过uv run前缀执行uv run python server.pyuv 会自动复用uv sync创建的同一虚拟环境。3.2 配置环境变量原始文档明确要求设置两个环境变量LINKUP_API_KEY... OPENAI_API_KEY...LINKUP_API_KEY必填由 server.py 的LinkupClient()读取用于网页搜索鉴权。原始文档提示到 Linkup 官网申请。OPENAI_API_KEYREADME 要求一并配置。需要说明的是从当前仓库源码看server.py 与 rag.py 中并未直接 import openai——RAG 生成端实际走的是本地 Ollama。可以推断该变量更多是原始教程基于 OpenAI 后端方案的通用要求若你完全复刻本仓库的本地链路务必保证至少LINKUP_API_KEY有效同时保留OPENAI_API_KEY以备后续切换云端 LLM 或遵循视频原流程亦不会造成冲突。配置文件加载由 server.py 顶部的load_dotenv()完成因此你也可以在项目根目录放一个.env文件来存放以上密钥。3.3 准备 RAG 数据RAG 工具读取的是启动时传入的目录。仓库内置了示例文档data/DeepSeek.pdf关于 DeepSeek 模型的中文/技术资料无需额外准备即可跑通若要测试自己的私有资料将 PDF或 LlamaIndexSimpleDirectoryReader支持的其他格式放入同一目录即可。4. MCP 服务器入口 server.py 逐行拆解server.py 是整个项目的核心入口逻辑非常紧凑——通过 FastMCP 只做三件事初始化、注册工具、启动。4.1 初始化from dotenv import load_dotenv from linkup import LinkupClient from rag import RAGWorkflow from mcp.server.fastmcp import FastMCP load_dotenv() mcp FastMCP(linkup-server) client LinkupClient() rag_workflow RAGWorkflow()FastMCP(linkup-server)创建命名 MCP 服务器。工具名、描述会以 MCP 协议暴露给 Cursor供其 Agent 决定何时调用。LinkupClient()Linkup SDK 的无参客户端API Key 从环境变量注入。RAGWorkflow()实例化 rag.py 中定义的 LlamaIndex Workflow此时尚未灌入文档索引为None。4.2 工具一Linkup 深度网页搜索mcp.tool() def web_search(query: str) - str: Search the web for the given query. search_response client.search( queryquery, depthstandard, # standard or deep output_typesourcedAnswer, # searchResults or sourcedAnswer or structured structured_output_schemaNone, # must be filled if output_type is structured ) return search_responsemcp.tool()装饰器会把下方函数自动声明为 MCP 工具Cursor 侧即可在对话中调用名为web_search、签名含query参数的工具。核心参数query搜索问题由调用方Cursor 的 Agent根据用户意图自动生成。depthstandard搜索深度枚举可选standard或deep。deep模式会执行更深层级的抓取与推理通常带来更全面结果但耗时更长standard适合日常快速检索。output_typesourcedAnswer输出类型三选一searchResults返回原始检索结果列表sourcedAnswer由 Linkup 基于检索结果综合生成答案并附带出处本项目的默认选择兼顾回答质量与可溯源structured返回结构化数据。structured_output_schemaNone仅当output_typestructured时必须提供 schema源码注释明确提示 must be filled if output_type is structured否则保持None。函数 docstringSearch the web for the given query.会作为工具描述同步进 Cursor 的能力列表帮助模型判断什么时候该调用网页搜索因此写好一句话描述对 Agent 命中率很重要。4.3 工具二本地文档 RAGmcp.tool() async def rag(query: str) - str: Use a simple RAG workflow to answer queries using documents from data directory about Deep Seek response await rag_workflow.query(query) return str(response)rag是异步工具内部直接委托给rag_workflow.query(query)。工具描述明确告知模型它回答的是data目录中关于 DeepSeek的文档问题——这一定位让 Cursor 能自发区分该联网搜调 web_search还是该查本地资料调 rag。4.4 启动入口if __name__ __main__: asyncio.run(rag_workflow.ingest_documents(data)) mcp.run(transportstdio)启动序列值得细看先用asyncio.run(rag_workflow.ingest_documents(data))预灌入文档——读取data目录全部文件并构建向量索引失败则后续 rag 查询会报 No documents have been ingested再以mcp.run(transportstdio)启动服务器。stdio 传输是 MCP 客户端Cursor、Claude Desktop 等以子进程方式拉起服务器的标准通道Cursor 启动该命令后通过标准输入/输出与该进程通信。5. rag.py 源码剖析LlamaIndex 事件驱动 RAG 工作流如果说 server.py 是协议层rag.py 就是引擎层。它不是简单的query_engine一行调用而是用 LlamaIndex 的新一代事件驱动 Workflow将 RAG 拆成可观察、可编排的三个步骤。5.1 事件与工作流骨架nest_asyncio.apply() class RetrieverEvent(Event): Result of running retrieval nodes: list[NodeWithScore] class RAGWorkflow(Workflow): def __init__(self, model_namellama3.2, embedding_modelBAAI/bge-small-en-v1.5): super().__init__() self.llm Ollama(modelmodel_name) self.embed_model HuggingFaceEmbedding(model_nameembedding_model) Settings.llm self.llm Settings.embed_model self.embed_model self.index None关键点RAGWorkflow继承llama_index.core.workflow.Workflow用step装饰的异步方法构成有向图步骤间通过自定义事件如RetrieverEvent传递数据。模型默认值写死在构造参数上LLM 为经 Ollama 调用的llama3.2embedding 为本地加载的BAAI/bge-small-en-v1.5HuggingFace 小模型CPU 即可运行。通过Settings.llm/Settings.embed_model将模型写入 LlamaIndex 全局设置后续索引与合成器会自动继承。nest_asyncio.apply()允许嵌套事件循环。这是本地 Jupyter/混合调度场景的经典坑——当外层已有事件循环如 FastMCP 的 async 运行环境时内层再次asyncio.run会冲突nest_asyncio正是为此打补丁。5.2 三个 stepingest → retrieve → synthesize步骤一灌入文档ingeststep async def ingest(self, ctx: Context, ev: StartEvent) - StopEvent | None: dirname ev.get(dirname) if not dirname: return None documents SimpleDirectoryReader(dirname).load_data() self.index VectorStoreIndex.from_documents(documentsdocuments) return StopEvent(resultself.index)用SimpleDirectoryReader加载目录内文档VectorStoreIndex.from_documents在内存中完成切块 embedding 建索引。索引暂存于self.index注意内存索引进程重启后需重新灌入这正是 server.py 每次启动先 ingest 的原因。步骤二检索retrievestep async def retrieve(self, ctx: Context, ev: StartEvent) - RetrieverEvent | None: query ev.get(query) index ev.get(index) or self.index if not query: return None if index is None: print(Index is empty, load some documents before querying!) return None retriever index.as_retriever(similarity_top_k2) nodes await retriever.aretrieve(query) await ctx.set(query, query) return RetrieverEvent(nodesnodes)index.as_retriever(similarity_top_k2)取 top-2 最相关文本块similarity_top_k直接控制送入 LLM 的上下文量可按知识密度上调。空索引保护未灌文档时打印提示并优雅返回None避免下游空指针。ctx.set(query, query)把用户问题写入步骤共享的Context供后续合成步骤读取。步骤三合成回答synthesizestep async def synthesize(self, ctx: Context, ev: RetrieverEvent) - StopEvent: summarizer CompactAndRefine(streamingTrue, verboseTrue) query await ctx.get(query, defaultNone) response await summarizer.asynthesize(query, nodesev.nodes) return StopEvent(resultresponse)使用 LlamaIndex 的CompactAndRefine合成器流式输出、verbose 打印中间过程CompactAndRefine会先尽量把检索块压缩进单次 prompt超出窗口时再采用逐块精炼refine策略累加答案是长上下文场景下的稳健默认选择。5.3 对外封装方法async def query(self, query_text: str): if self.index is None: raise ValueError(No documents have been ingested. Call ingest_documents first.) result await self.run(queryquery_text, indexself.index) return result async def ingest_documents(self, directory: str): result await self.run(dirnamedirectory) self.index result return resultquery/ingest_documents是给 server.py 与独立测试用的门面方法前者把用户问题以queryquery_text, indexself.index塞进StartEvent启动整条流水线后者把目录路径塞进StartEvent触发灌入。rag.py文件底部还附带了独立演示入口main()可脱离 MCP 直接验证 RAG 链路workflow RAGWorkflow() await workflow.ingest_documents(data) result await workflow.query(How was DeepSeekR1 trained?) async for chunk in result.async_response_gen(): print(chunk, end, flushTrue)6. 在 Cursor 中接入并运行 MCP 服务器6.1 以 stdio 子进程方式注册server.py 以transportstdio启动意味着 MCP 客户端需要把该命令当作子进程拉起。在 Cursor 的 MCP 服务器配置中添加一条本地命令型服务器对应type: stdio启动命令可参考uv run python server.py前提是命令的工作目录指向仓库根目录保证能 importrag模块、能加载data目录、能读取.env。Cursor 在识别到该服务器提供的web_search、rag两个工具后便可在对话中按需调用。补充说明mcp[cli]1.5.0依赖同时带入了 MCP 官方命令行工具如mcp run/mcp dev等可在接入 Cursor 前用其对server.py做协议级冒烟测试属官方 SDK 自带的排查手段本项目依赖声明已为其预留了安装条件。6.2 启动后的典型运行链路服务真正跑起来后一次典型问答会这样流转Cursor 的 Agent 判断用户问题需要实时信息 → 调用web_search服务器内LinkupClient.search(query, depth, output_type)发起联网检索返回带出处的综合答案若用户问题指向本地 DeepSeek 资料 → 调用rag→RAGWorkflow.run走retrieve(similarity_top_k2) → CompactAndRefine 合成流水线基于DeepSeek.pdf作答。7. 运行注意要点与排错清单结合源码实现落地时需重点检查以下事项现象排查方向依据rag返回 No documents have been ingestedserver.py主流程必须在mcp.run前完成 ingest确认data目录非空rag.py日志打印 Index is empty...检索前索引为空灌入失败或被跳过rag.pyRAG 调用报事件循环冲突确认nest_asyncio.apply()在导入即执行模块顶层已调用rag.pyRAG 无响应/报连接错误本地 Ollama 服务需运行且已ollama pull llama3.2模型名可在RAGWorkflow(model_name...)处调整rag.pyweb_search报鉴权错误LINKUP_API_KEY未配置或.env未加载load_dotenv()在模块顶层执行server.py需要结构化输出但报错output_typestructured时必须同时传structured_output_schemaserver.py换目录后 RAG 仍答旧文档索引为进程内存态重启进程重新 ingest使新文档生效server.py8. 小结这套架构的可迁移价值从本仓库可以抽象出一个可复用的 MCP Agent 工具接入范式协议层FastMCP 工具注册 能力层外部 API / 本地 RAG 引擎分层解耦。其中 server.py 展示的是一行装饰器 一个 Cursor 可用工具的最小接入成本rag.py 展示的是把 RAG 显式建模为 ingest / retrieve / synthesize 事件流、从而让每一步可观测可替换的思路。想要扩展其他能力时只需沿用mcp.tool()追加新函数并把新引擎以同样的 Workflow 或 client 对象注入即可——这正是 MCP 生态下为 AI 编码工具持续插拔能力的标准姿势。【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考