企业官网建设流程全解析

手把手为 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）的能力边界往往取决于其能否"触达"外部数据：实时的网页信息、私有的本地文档。本项目的目标非常聚焦——构建一个自定义 MCP（Model Context Protocol）服务器，把它接入 Cursor，让 Cursor 获得两种此前不具备的能力：

1. 深度网页搜索（Web Search）：通过 Linkup 服务执行实时联网搜索，并返回带出处的结构化答案；
2. 本地文档 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 = [ "ipykernel>=6.29.5", "linkup-sdk>=0.2.4", "llama-index>=0.12.25", "llama-index-embeddings-huggingface>=0.5.2", "llama-index-llms-ollama>=0.5.3", "mcp[cli]>=1.5.0", ]

各依赖在本文架构中的角色如下：

两个前置环境约束值得注意：其一，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.toml+uv.lock创建虚拟环境并安装全部依赖。因为代码已提交了uv.lock，同步结果具备确定性；如需后续运行脚本，建议通过uv run前缀执行（uv run python server.py），uv 会自动复用uv sync创建的同一虚拟环境。

3.2 配置环境变量

原始文档明确要求设置两个环境变量：

LINKUP_API_KEY=... OPENAI_API_KEY=...

• LINKUP_API_KEY：必填，由 server.py 的LinkupClient()读取，用于网页搜索鉴权。原始文档提示到 Linkup 官网申请。
• OPENAI_API_KEY：README 要求一并配置。需要说明的是，从当前仓库源码看，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( query=query, depth="standard", # "standard" or "deep" output_type="sourcedAnswer", # "searchResults" or "sourcedAnswer" or "structured" structured_output_schema=None, # must be filled if output_type is "structured" ) return search_response

@mcp.tool()装饰器会把下方函数自动声明为 MCP 工具，Cursor 侧即可在对话中调用名为web_search、签名含query参数的工具。核心参数：

• query：搜索问题，由调用方（Cursor 的 Agent）根据用户意图自动生成。
• depth="standard"：搜索深度枚举，可选"standard"或"deep"。deep模式会执行更深层级的抓取与推理，通常带来更全面结果但耗时更长；standard适合日常快速检索。
• output_type="sourcedAnswer"：输出类型三选一：
  • searchResults：返回原始检索结果列表；
  • sourcedAnswer：由 Linkup 基于检索结果综合生成答案，并附带出处（本项目的默认选择，兼顾回答质量与可溯源）；
  • structured：返回结构化数据。
• structured_output_schema=None：仅当output_type="structured"时必须提供 schema（源码注释明确提示 "must be filled if output_type is 'structured'"），否则保持None。

函数 docstringSearch the web for the given query.会作为工具描述同步进 Cursor 的能力列表，帮助模型判断"什么时候该调用网页搜索"，因此写好一句话描述对 Agent 命中率很重要。

4.3 工具二：本地文档 RAG

@mcp.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(transport="stdio")

启动序列值得细看：

1. 先用asyncio.run(rag_workflow.ingest_documents("data"))预灌入文档——读取data目录全部文件并构建向量索引（失败则后续 rag 查询会报 "No documents have been ingested"）；
2. 再以mcp.run(transport="stdio")启动服务器。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_name="llama3.2", embedding_model="BAAI/bge-small-en-v1.5"): super().__init__() self.llm = Ollama(model=model_name) self.embed_model = HuggingFaceEmbedding(model_name=embedding_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.2，embedding 为本地加载的BAAI/bge-small-en-v1.5（HuggingFace 小模型，CPU 即可运行）。
• 通过Settings.llm/Settings.embed_model将模型写入 LlamaIndex 全局设置，后续索引与合成器会自动继承。
• nest_asyncio.apply()：允许嵌套事件循环。这是本地 Jupyter/混合调度场景的经典坑——当外层已有事件循环（如 FastMCP 的 async 运行环境）时，内层再次asyncio.run会冲突，nest_asyncio正是为此打补丁。

5.2 三个 @step：ingest → retrieve → synthesize

步骤一：灌入文档（ingest）

@step 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(documents=documents) return StopEvent(result=self.index)

用SimpleDirectoryReader加载目录内文档，VectorStoreIndex.from_documents在内存中完成切块 + embedding + 建索引。索引暂存于self.index（注意：内存索引，进程重启后需重新灌入，这正是 server.py 每次启动先 ingest 的原因）。

步骤二：检索（retrieve）

@step 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_k=2) nodes = await retriever.aretrieve(query) await ctx.set("query", query) return RetrieverEvent(nodes=nodes)

• index.as_retriever(similarity_top_k=2)：取 top-2 最相关文本块（similarity_top_k直接控制送入 LLM 的上下文量，可按知识密度上调）。
• 空索引保护：未灌文档时打印提示并优雅返回None，避免下游空指针。
• ctx.set("query", query)：把用户问题写入步骤共享的Context，供后续合成步骤读取。

步骤三：合成回答（synthesize）

@step async def synthesize(self, ctx: Context, ev: RetrieverEvent) -> StopEvent: summarizer = CompactAndRefine(streaming=True, verbose=True) query = await ctx.get("query", default=None) response = await summarizer.asynthesize(query, nodes=ev.nodes) return StopEvent(result=response)

使用 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(query=query_text, index=self.index) return result async def ingest_documents(self, directory: str): result = await self.run(dirname=directory) self.index = result return result

query/ingest_documents是给 server.py 与独立测试用的"门面方法"：前者把用户问题以query=query_text, index=self.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="", flush=True)

6. 在 Cursor 中接入并运行 MCP 服务器

6.1 以 stdio 子进程方式注册

server.py 以transport="stdio"启动，意味着 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 启动后的典型运行链路

服务真正跑起来后，一次典型问答会这样流转：

1. Cursor 的 Agent 判断用户问题需要实时信息 → 调用web_search；
2. 服务器内LinkupClient.search(query, depth, output_type)发起联网检索，返回带出处的综合答案；
3. 若用户问题指向本地 DeepSeek 资料 → 调用rag→RAGWorkflow.run走retrieve(similarity_top_k=2) → CompactAndRefine 合成流水线，基于DeepSeek.pdf作答。

7. 运行注意要点与排错清单

结合源码实现，落地时需重点检查以下事项：

8. 小结：这套架构的可迁移价值

从本仓库可以抽象出一个可复用的 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