1. 为什么我要自己写 MCP Server 而不是接第三方先说清楚这套东西是什么。MCP 是 Model Context Protocol你可以把它理解成“大模型调用外部工具的统一插座标准”模型不直接碰你的数据库、文件、搜索引擎而是通过一个 Server 暴露出来的工具Tool去调用。Agentic RAG 则是让 Agent 自己决定“这个问题该查哪个索引、要不要再搜一下网络、要不要先做摘要”而不是固定一条检索链路走到底。这套组合适合谁适合已经会写 Python、想让本地文档问答从“一次性脚本”升级成“可维护系统”的开发者也适合被第三方 MCP Server 的权限和黑盒行为卡住、想自己掌控检索链路的人。我踩过的第一个坑就是一开始图省事直接用了别人封装好的 MCP Server。问题很快暴露它的工具粒度是写死的我想加一个“按文档摘要回答”的管道只能改它的源码它的缓存策略我看不到同一份 PDF 被反复解析、反复 embedding账单肉眼可见地涨更麻烦的是它把模型调用地址也一起封装了我想换一个统一的 Key 通道得翻好几层配置。于是我决定MCP Server 自己写工具边界自己定模型调用统一走一个 Key 通道。这里就引出本文的第二个主角——TaoToken。它做的事情很朴素给你一个统一的 API 通道和一把 KeyOpenAI 兼容格式Base URL 是https://taotoken.net/api。我的 MCP Server 里做 embedding、做摘要、做 Agent 推理全都指向这一个地址不用在 LlamaIndex、LangGraph、各个 SDK 之间来回配不同的 Key。对自建 MCP 架构来说这一点很关键Server 端和 Client 端是两个进程如果它们各自维护一套模型凭证排障会非常痛苦。统一通道之后我只需要在一个地方管 Key。所以整篇文章的目标很明确不依赖任何第三方 MCP Server从零把 MCP ServerRAG 管道 MCP ClientLangGraph Agent搭起来模型调用统一走 TaoToken最后跑通“本地 MCP 工具调用 RAG 问答闭环”。下面按“先讲架构分工再上可复制配置再验证再排错”的顺序来你可以跟着一步步敲。2. TaoToken 统一 Key 接入 MCP 架构的前置准备在动手写 Server 之前先把“模型从哪来”这件事定死否则后面代码里到处是硬编码的 Key改起来想砸键盘。TaoToken 在这里扮演的是统一模型入口MCP Server 里的 embedding 模型、Client 里的对话/推理模型都通过它调用。你需要先拿到一把 API Key入口在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。拿到之后不要写进代码放进环境变量。我建议在项目根目录建一个.env内容大致是这样把sk-xxx换成你自己的# .env TAOTOKEN_API_KEYsk-xxx TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里统一读取。这里有个细节LlamaIndex 和 LangGraph 底层大多走 OpenAI 兼容接口所以只要把base_url和api_key指对模型名按你实际开通的填即可。下面是我封装的一个最小模型工厂Server 和 Client 共用# llm_factory.py import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() def get_client() - OpenAI: return OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], # https://taotoken.net/api ) # 对话/推理模型名按你实际开通的填 CHAT_MODEL gpt-4o-mini # embedding 模型名按你实际开通的填 EMBED_MODEL text-embedding-3-small为什么强调“统一”因为 MCP 架构天然是 Client/Server 分离的。Server 端要调 embedding 建索引Client 端要调对话模型做 Agent 推理。如果两边各配一套凭证一旦某次请求 401你得先判断是 Server 的 Key 过期还是 Client 的 Key 写错。统一到 TaoToken 之后401 基本只有一个原因这把 Key 本身有问题排查范围直接砍半。前置准备还包括依赖安装。我用的技术栈是Server 端 LlamaIndex Chroma 做 RAG 管道Client 端 LangGraph 做 AgentMCP 通信用 SSE 模式。安装命令pip install mcp[cli] llama-index llama-index-vector-stores-chroma \ chromadb langgraph langchain-openai python-dotenv装完之后先别急着写业务代码跑一个最小连通性测试确认 TaoToken 通道是通的# smoke_test.py from llm_factory import get_client, CHAT_MODEL client get_client() resp client.chat.completions.create( modelCHAT_MODEL, messages[{role: user, content: 只回复两个字通了}], ) print(resp.choices[0].message.content)能打印出“通了”说明 Key 和 Base URL 都没问题可以进入下一步。如果这一步就报错先去看第 5 节的排错对照表别往下硬写。3. 可复制的 MCP Server 与 Client 配置片段这一节是全文最“抄了就能用”的部分。我先把两个配置文件摆出来再讲 Server 工具和 Client Agent 的关键代码。配置文件路径和字段名请保持一致后面排错时对得上。3.1 mcp_config.jsonClient 连接 Server 的配置这个文件放在 Client 项目根目录描述要连哪些 MCP Server、用哪种 transport、允许加载哪些工具{ servers: { rag_server: { transport: sse, url: http://localhost:5050/sse, allowed_tools: [ create_vector_index, query_document, get_document_summary, list_indexes ] } } }注意allowed_tools这个字段它是我在基础 MCP 客户端上扩展出来的工具白名单。为什么要白名单因为 Agent 拿到工具列表后会自己推理该调哪个如果 Server 暴露了“删除索引”这类危险工具模型有可能在你不希望的时候调用它。白名单把 Agent 能看到的工具收窄等于给它划了活动范围。3.2 doc_config.json知识文档与索引参数配置这个文件描述“有哪些文档、各自对应哪个索引、切块参数是多少”。它会在构建 Agent 时被注入系统提示词让模型知道有哪些索引名可用{ data/c-rag.pdf: { description: c-rag 技术论文回答 c-rag 相关问题, index_name: c-rag, chunk_size: 500, chunk_overlap: 50 }, data/questions.csv: { description: 税务问题数据集包含常见咨询问答, index_name: tax-questions, chunk_size: 500, chunk_overlap: 50 } }3.3 MCP Servercreate_vector_index 工具Server 端我用 LlamaIndex 实现 RAG 管道用 Chroma 做向量库。核心工具create_vector_index带缓存逻辑文档内容 hash 切块参数组成缓存名只要这两者不变就不重复解析、不重复 embedding。下面是可以直接跑的版本# rag_server.py import os import hashlib from mcp.server.fastmcp import FastMCP, Context from llama_index.core import VectorStoreIndex, StorageContext from llama_index.vector_stores.chroma import ChromaVectorStore from llama_index.core.node_parser import SentenceSplitter from llm_factory import get_client, EMBED_MODEL app FastMCP(rag_server) STORAGE_DIR ./storage CACHE_DIR ./cache os.makedirs(STORAGE_DIR, exist_okTrue) os.makedirs(CACHE_DIR, exist_okTrue) def get_cache_path(file_path: str, chunk_size: int, chunk_overlap: int) - str: with open(file_path, rb) as f: content_hash hashlib.md5(f.read()).hexdigest() name f{os.path.basename(file_path)}_{content_hash}_{chunk_size}_{chunk_overlap} return os.path.join(CACHE_DIR, name) app.tool() async def create_vector_index( ctx: Context, file_path: str, index_name: str, chunk_size: int 500, chunk_overlap: int 50, force_recreate: bool False, ) - str: 创建或加载文档向量索引带缓存避免重复解析与嵌入 storage_path os.path.join(STORAGE_DIR, index_name) cache_path get_cache_path(file_path, chunk_size, chunk_overlap) need_recreate ( force_recreate or not os.path.exists(storage_path) or not os.path.exists(cache_path) ) if os.path.exists(storage_path) and not need_recreate: return f索引 {index_name} 已存在且参数未变化无需创建 chroma ctx.request_context.lifespan_context.chroma try: chroma.delete_collection(nameindex_name) except Exception: pass # 首次创建时集合不存在忽略 collection chroma.get_or_create_collection(nameindex_name) vector_store ChromaVectorStore(chroma_collectioncollection) storage_context StorageContext.from_defaults(vector_storevector_store) from llama_index.core import SimpleDirectoryReader docs SimpleDirectoryReader(input_files[file_path]).load_data() splitter SentenceSplitter(chunk_sizechunk_size, chunk_overlapchunk_overlap) nodes splitter.get_nodes_from_documents(docs) VectorStoreIndex( nodes, storage_contextstorage_context, embed_modelfopenai:{EMBED_MODEL}, ) # 写入缓存标记下次命中即跳过 with open(cache_path, w) as f: f.write(index_name) return f成功创建索引: {index_name}, 包含 {len(nodes)} 个节点这里有个容易忽略的点embed_model我写的是openai:{EMBED_MODEL}LlamaIndex 会走 OpenAI 兼容协议而 OpenAI 客户端的 base_url 需要指向 TaoToken。最稳妥的做法是在 Server 启动时设置环境变量OPENAI_API_KEY和OPENAI_BASE_URL让底层 SDK 自动读取# server_main.py import os os.environ[OPENAI_API_KEY] os.environ[TAOTOKEN_API_KEY] os.environ[OPENAI_BASE_URL] os.environ[TAOTOKEN_BASE_URL]3.4 MCP ClientLangGraph Agent 构建Client 端用 LangGraph 的create_react_agent快速搭一个 ReAct Agent工具列表从 MCP Server 动态拉取# rag_agent.py import json from langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI from llm_factory import CHAT_MODEL SYSTEM_PROMPT 你是一个文档问答助手。可用索引如下 {doc_info_str} 规则事实性问题用 query_document总结性问题用 get_document_summary 需要实时信息时调用搜索工具。当前时间{current_time} async def build_agent(mcp_client, doc_config: dict): mcp_tools await mcp_client.get_tools_for_langgraph() doc_info_str \n.join( f- {path}: {cfg[description]} (index_name{cfg[index_name]}) for path, cfg in doc_config.items() ) llm ChatOpenAI( modelCHAT_MODEL, base_urlos.environ[TAOTOKEN_BASE_URL], api_keyos.environ[TAOTOKEN_API_KEY], ) return create_react_agent( modelllm, toolsmcp_tools, promptSYSTEM_PROMPT.format( doc_info_strdoc_info_str, current_timedatetime.now().strftime(%Y-%m-%d %H:%M:%S), ), )三件套在这里齐了Base URL 是https://taotoken.net/apiKey 来自环境变量Model ID 是CHAT_MODEL。Server 端 embedding 同理只是 Model ID 换成EMBED_MODEL。把这三样对齐模型调用就不会出岔子。4. 端到端验证从启动 Server 到 RAG 问答闭环配置写完接下来是验证动作。我把它拆成四步每步都有明确的“成功信号”你对照着看就知道卡在哪。第一步启动 MCP Server。SSE 模式下 Server 会监听一个端口启动命令python server_main.py成功信号终端打印出工具清单能看到create_vector_index、query_document、get_document_summary、list_indexes四个工具名并且提示 SSE 服务已在http://localhost:5050/sse就绪。如果工具清单是空的说明app.tool()装饰器没生效检查 FastMCP 版本。第二步准备文档和配置。把要索引的 PDF、CSV 放进data/目录确认doc_config.json里的路径和实际文件名一致。这一步不做任何预处理解析和切块都由 Server 工具完成。第三步启动 Client观察首次运行日志python rag_agent_langgraph.py首次运行会看到Client 连接 Server → 调用create_vector_index逐个建索引 → 加载工具 → 构建 Agent。因为缓存目录是空的每个文档都会被解析和 embedding日志里会打印“成功创建索引: xxx, 包含 N 个节点”。这一步耗时取决于文档大小耐心等。第四步退出程序再启动一次验证缓存生效。第二次启动时日志应该显示“索引 xxx 已存在且参数未变化无需创建”并且几乎瞬间完成。这就是缓存机制在起作用——文档内容 hash 和切块参数都没变Server 直接跳过解析和嵌入。如果你改了 PDF 内容但文件名没变hash 会变缓存失效索引自动重建这正是我想要的行为。第五步进入交互式问答测三类问题。第一类事实性查询“北京和上海的城市信息分别是什么”日志里应该看到 Agent 分别调用了两个索引的query_document甚至可能额外调用搜索工具补充。第二类总结性问题“帮我总结一下这份税务数据集的主要内容。”这时 Agent 应该走get_document_summary而不是向量检索。第三类索引管理“把 csv 文档的索引重建一下。”Agent 会推理出create_vector_index并带上force_recreatetrue参数。成功信号很直观Agent 的回答里引用了文档内容且日志显示工具调用链路符合预期。到这一步本地 MCP 工具调用 RAG 问答闭环就跑通了。如果你想在浏览器里直接对比模型输出、确认通道没问题可以打开模型对话页面 https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 手动问一句和 Agent 的回答对照着看。5. 本篇常见报错排查对照表这一节按真实报错来我把搭这套系统时遇到的坑列成对照表你遇到问题时直接查。报错/现象可能原因排查动作401 UnauthorizedKey 没读到或写错检查.env是否被load_dotenv()加载确认TAOTOKEN_API_KEY无多余空格Server 端是否设置了OPENAI_API_KEYlocal proxy failed/ 连接被拒Base URL 写错或网络不通确认地址是https://taotoken.net/api不要漏/api先用smoke_test.py单独验证通道Error reading choices/ 返回体解析失败模型名不存在或返回了非预期结构确认CHAT_MODEL、EMBED_MODEL是你实际开通的模型 ID打印原始resp看返回内容OAuth相关报错误用了需要 OAuth 的接入方式本文走的是 API Key 方式不需要 OAuth检查是否混入了其他 SDK 的认证逻辑SSE 连接超时Server 没启动或端口被占curl http://localhost:5050/sse看是否有响应换端口重试工具清单为空app.tool()未生效确认 FastMCP 版本检查装饰器是否写在 async 函数上索引反复重建缓存路径不可写或 hash 变化检查CACHE_DIR权限确认文档内容确实没变Agent 不调用工具直接瞎答系统提示词没注入索引信息检查doc_info_str是否为空确认doc_config.json路径正确重点说两个高频的。第一个是 401十有八九是环境变量没加载。我建议在 Server 和 Client 的入口文件最顶部都加一句load_dotenv()并且打印一次os.environ.get(TAOTOKEN_BASE_URL)确认读到了。第二个是local proxy failed这个报错通常意味着请求根本没发到目标地址先确认 Base URL 拼写再用最小脚本验证别在业务代码里猜。还有一个隐蔽的坑Server 端和 Client 端如果用了不同的模型配置方式比如 Server 走环境变量、Client 走显式参数容易出现“一边通一边不通”。我的做法是两边都从同一个llm_factory.py读配置保证 Base URL、Key、Model ID 三件套完全一致。这样任何一边出问题另一边也能快速复现排查效率高很多。6. 长期跑 Agent 与 Coding 场景的接入建议这套系统跑通之后你会发现它不只是个文档问答 demo。MCP 架构带来的模块化让 Server 端可以独立扩展想加多模态解析改 Server想换向量库改 ServerClient 端的 Agent 完全不用动。这种松耦合在长期维护里价值很大。如果你打算把它用在日常编码或 Agent 工作流里有两点建议。第一把模型调用通道固定下来别今天用这个明天换那个否则每次换都要重新验证一遍链路。TaoToken 的 Coding Plan 适合这种长期、高频的调用场景入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 你可以按自己的调用量评估。第二接入文档建议通读一遍尤其是错误码和参数说明能省下大量试错时间https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。最后给一个实用技巧在 Server 端加一个list_indexes辅助工具返回当前所有索引名和对应文档。Agent 在推理时如果拿不准该查哪个索引可以先调这个工具确认再决定查询参数。这个小工具在文档数量多的时候特别有用能明显减少 Agent “猜错索引”的情况。代码很短照着create_vector_index的结构写一个只读工具即可注意别把它加进危险操作白名单之外的地方。