1. 为什么要把 Agentic RAG 拆成一个独立 server先说清楚这篇要解决的事你手里有一套知识库Dify 数据集也好、自建向量库也好现在想让 Dify、Cherry Studio 这类编排工具通过一个标准接口去调用它而不是把检索逻辑硬塞进每个 Agent 的提示词里。MCPModel Context Protocol就是干这个的——它把「检索」这件事声明成一个工具任何支持 MCP 的客户端都能发现并调用。Agentic RAG 和普通 RAG 的区别在于普通 RAG 是「先检索再回答」的固定流水线而 Agentic RAG 把检索当成 Agent 可以自主决定调用的工具什么时候查、查几次、用什么 query 都由模型判断。要做到这一点检索能力必须暴露成一个独立的、可被发现的 server而不是藏在某个函数里。这篇适合谁已经跑通 Dify 知识库、想把它接进 MCP 生态的开发者或者反过来手上有 MCP 客户端Claude Code、Cherry Studio 等想给它挂一个自己的检索后端。技术栈是 FastAPI MCP Python SDK全程可复制。我试过把检索逻辑直接写进 Agent 的 system prompt 里让它「假装调用」结果模型经常编造检索结果。拆成独立 server 后工具调用是真实的 HTTP 请求返回带引用的原文片段幻觉明显下降。2. 前置准备TaoToken 与项目骨架2.1 拿到调用凭证如果你还没配好模型侧的访问凭证先去 TaoToken 控制台创建一个 API Key。整个链路里MCP server 负责检索模型负责决策两者都需要各自的凭证。模型对话相关的 Key 在 TaoToken API Keys 页面创建接入细节看 接入文档。注意检索用的 Dify dataset token 和模型用的 API Key 是两套东西别混。前者在 Dify 知识库的「API 访问」里生成后者在 TaoToken 控制台生成。2.2 项目结构mcp-rag-server/ ├── main.py # FastAPI MCP 挂载 ├── rag.py # 检索工具实现 ├── config.py # 环境变量 └── requirements.txt依赖装这几个pip install fastapi uvicorn httpx mcp python-dotenvconfig.py把凭证收拢避免硬编码import os from dotenv import load_dotenv load_dotenv() DIFY_BASE os.getenv(DIFY_BASE, https://api.dify.ai/v1) DIFY_DATASET_ID os.getenv(DIFY_DATASET_ID) DIFY_DATASET_TOKEN os.getenv(DIFY_DATASET_TOKEN) TOP_K int(os.getenv(TOP_K, 3)).env里填DIFY_DATASET_ID你的数据集ID DIFY_DATASET_TOKENdataset-xxxxxxxx TOP_K33. 可复制配置MCP 工具声明与 FastAPI 挂载3.1 检索工具实现核心是把 Dify 的 retrieve 接口包成一个 MCP tool。注意mcp.tool()装饰器下的函数签名和 docstring 会被客户端读取docstring 写清楚模型才知道什么时候调它。# rag.py import logging import httpx from mcp.server.fastmcp import FastMCP from config import DIFY_BASE, DIFY_DATASET_ID, DIFY_DATASET_TOKEN, TOP_K logger logging.getLogger(__name__) mcp FastMCP(agentic-rag-server) _client httpx.AsyncClient(timeout30.0) mcp.tool() async def rag_retrieve(query: str) - dict: 检索产品业务知识库返回与 query 相关的原文片段及引用来源。 Args: query: 检索内容必传。例如 客服电话是多少 url f{DIFY_BASE}/datasets/{DIFY_DATASET_ID}/retrieve headers { Authorization: fBearer {DIFY_DATASET_TOKEN}, Content-Type: application/json, } payload { query: query, retrieval_model: { search_method: semantic_search, reranking_enable: False, top_k: TOP_K, score_threshold_enabled: False, }, } logger.info(rag_retrieve query%s, query) try: resp await _client.post(url, headersheaders, jsonpayload) resp.raise_for_status() data resp.json() except Exception as e: logger.error(rag_retrieve failed: %s, e) raise ValueError(f检索接口调用失败: {e}) records data.get(records, []) return { query: query, count: len(records), results: [ { content: r[segment][content], score: r.get(score), source: r[segment].get(document, {}).get(name, unknown), } for r in records ], }返回结构里带上source和score这是「带引用」的关键——模型拿到后可以在回答里标注来源用户也能核对。3.2 FastAPI 挂载 MCPMCP Python SDK 提供了 SSE 传输可以直接挂到 FastAPI 的 app 上这样检索 server 和普通 HTTP 服务共用一个端口。# main.py import logging from fastapi import FastAPI from rag import mcp logging.basicConfig(levellogging.INFO) app FastAPI(titleAgentic RAG MCP Server) # 把 MCP 的 SSE 应用挂到 /mcp 路径 app.mount(/mcp, mcp.sse_app()) app.get(/health) async def health(): return {status: ok} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动uvicorn main:app --host 0.0.0.0 --port 8000看到Uvicorn running on http://0.0.0.0:8000就说明起来了。MCP 的 SSE 端点在http://你的IP:8000/mcp/sse。3.3 Dify 侧接入配置Dify 里新建一个 Agent 应用在「工具」里添加 MCP 服务填 SSE 地址http://你的服务器IP:8000/mcp/sse保存后 Dify 会自动拉取工具列表应该能看到rag_retrieve。如果拉不到先确认/health能访问再确认 SSE 路径没写错。提示Dify 和 MCP server 不在同一台机器时注意防火墙放行 8000 端口且用内网 IP 或域名别用 localhost。4. 验证请求端到端跑一次检索4.1 直接测 MCP 工具先用 curl 确认 SSE 端点活着curl -N http://127.0.0.1:8000/mcp/sse正常会持续输出event: endpoint之类的心跳。然后写个最小客户端脚本模拟 MCP 客户端调用工具# test_client.py import asyncio from mcp import ClientSession from mcp.client.sse import sse_client async def main(): async with sse_client(http://127.0.0.1:8000/mcp/sse) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具:, [t.name for t in tools.tools]) result await session.call_tool( rag_retrieve, {query: 客服电话是多少} ) print(检索结果:, result.content) asyncio.run(main())跑起来应该看到可用工具: [rag_retrieve] 检索结果: [{type: text, text: {query: 客服电话是多少, count: 2, results: [...]}}]count大于 0 且results里有content和source说明请求命中了知识库并返回了带引用的结果。4.2 在 Dify 里验证回到 Dify 的 Agent输入「帮我查一下客服电话」观察运行日志。正常流程是模型判断需要检索 → 调用rag_retrieve→ 拿到片段 → 生成带来源的回答。如果模型没调工具检查工具描述是否够清晰或者手动在提示词里加一句「涉及业务信息时优先调用 rag_retrieve」。4.3 用模型对话侧验证如果你想让模型侧也走统一入口可以在 TaoToken 模型对话 里配一个支持工具调用的模型把 MCP server 挂上去做联调。长期跑编码或 Agent 任务的话Coding Plan 更适合持续调用场景。5. 本篇常见错排查报错一httpx.ConnectError: All connection attempts failedDify 地址写错或网络不通。先curl一下https://api.dify.ai/v1/datasets看能不能通。自建 Dify 的话把DIFY_BASE换成你的域名。报错二401 Unauthorizeddataset token 错了或过期。注意 token 前缀是dataset-别把应用的 API Key 填进来。重新在 Dify 知识库「API 访问」里生成。报错三404 Not Foundon/retrieveDIFY_DATASET_ID不对。这个 ID 在知识库 URL 里形如/datasets/xxxx-xxxx/documents取中间那段。报错四Dify 拉不到工具列表SSE 路径写成/mcp而不是/mcp/sse。SDK 的sse_app()默认挂在/sse子路径下所以完整地址是/mcp/sse。报错五检索返回count: 0知识库没内容或者search_method和你的索引类型不匹配。语义检索需要向量索引如果只建了关键词索引改成keyword_search或hybrid_search。报错六模型不调用工具docstring 太模糊。把「检索产品业务知识库」写具体比如「查询客服电话、退换货政策、产品参数等业务信息」模型判断触发时机的准确率会高很多。报错七返回内容乱码Dify 返回的是 UTF-8如果终端显示乱码是本地编码问题不影响实际数据。在代码里resp.json()拿到的就是正常字符串。6. 把检索能力沉淀成可复用资产这套结构跑通后你会发现它的价值不在「能查一次」而在于检索逻辑和编排逻辑彻底解耦了。同一个 MCP server 可以同时挂给 Dify、Cherry Studio、Claude Code甚至你自己写的 Agent 框架改检索策略只需要动rag.py一个文件。几个可以继续做的方向给rag_retrieve加一个top_k参数让模型自己控制召回数量加一个list_datasets工具支持多知识库路由把score_threshold打开过滤低分片段。这些都是在现有骨架上加一个mcp.tool()的事。凭证和接入配置统一在 TaoToken 控制台 管理需要新建 Key 或查用量时从那里进。检索 server 本身不依赖特定模型换模型不用改代码这是拆成独立 server 最实在的好处。