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 通道和一把 Key，OpenAI 兼容格式，Base URL 是https://taotoken.net/api。我的 MCP Server 里做 embedding、做摘要、做 Agent 推理，全都指向这一个地址，不用在 LlamaIndex、LangGraph、各个 SDK 之间来回配不同的 Key。对自建 MCP 架构来说，这一点很关键：Server 端和 Client 端是两个进程，如果它们各自维护一套模型凭证，排障会非常痛苦。统一通道之后，我只需要在一个地方管 Key。
所以整篇文章的目标很明确：不依赖任何第三方 MCP Server，从零把 MCP Server（RAG 管道）+ MCP Client（LangGraph 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_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。拿到之后不要写进代码，放进环境变量。
我建议在项目根目录建一个.env，内容大致是这样（把sk-xxx换成你自己的）：
# .env TAOTOKEN_API_KEY=sk-xxx TAOTOKEN_BASE_URL=https://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_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.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 做 Agent，MCP 通信用 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( model=CHAT_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.json：Client 连接 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 Server：create_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_ok=True) os.makedirs(CACHE_DIR, exist_ok=True) 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(name=index_name) except Exception: pass # 首次创建时集合不存在，忽略 collection = chroma.get_or_create_collection(name=index_name) vector_store = ChromaVectorStore(chroma_collection=collection) storage_context = StorageContext.from_defaults(vector_store=vector_store) from llama_index.core import SimpleDirectoryReader docs = SimpleDirectoryReader(input_files=[file_path]).load_data() splitter = SentenceSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap) nodes = splitter.get_nodes_from_documents(docs) VectorStoreIndex( nodes, storage_context=storage_context, embed_model=f"openai:{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 Client：LangGraph 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( model=CHAT_MODEL, base_url=os.environ["TAOTOKEN_BASE_URL"], api_key=os.environ["TAOTOKEN_API_KEY"], ) return create_react_agent( model=llm, tools=mcp_tools, prompt=SYSTEM_PROMPT.format( doc_info_str=doc_info_str, current_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ), )三件套在这里齐了：Base URL 是https://taotoken.net/api，Key 来自环境变量，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_recreate=true参数。
成功信号很直观：Agent 的回答里引用了文档内容，且日志显示工具调用链路符合预期。到这一步，本地 MCP 工具调用 + RAG 问答闭环就跑通了。如果你想在浏览器里直接对比模型输出、确认通道没问题，可以打开模型对话页面 https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 手动问一句，和 Agent 的回答对照着看。
5. 本篇常见报错排查对照表
这一节按真实报错来，我把搭这套系统时遇到的坑列成对照表，你遇到问题时直接查。
| 报错/现象 | 可能原因 | 排查动作 |
|---|---|---|
401 Unauthorized | Key 没读到或写错 | 检查.env是否被load_dotenv()加载；确认TAOTOKEN_API_KEY无多余空格；Server 端是否设置了OPENAI_API_KEY |
local 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；想换向量库，改 Server；Client 端的 Agent 完全不用动。这种松耦合在长期维护里价值很大。
如果你打算把它用在日常编码或 Agent 工作流里，有两点建议。第一，把模型调用通道固定下来，别今天用这个明天换那个，否则每次换都要重新验证一遍链路。TaoToken 的 Coding Plan 适合这种长期、高频的调用场景，入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ，你可以按自己的调用量评估。第二，接入文档建议通读一遍，尤其是错误码和参数说明，能省下大量试错时间：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
最后给一个实用技巧：在 Server 端加一个list_indexes辅助工具，返回当前所有索引名和对应文档。Agent 在推理时如果拿不准该查哪个索引，可以先调这个工具确认，再决定查询参数。这个小工具在文档数量多的时候特别有用，能明显减少 Agent “猜错索引”的情况。代码很短，照着create_vector_index的结构写一个只读工具即可，注意别把它加进危险操作白名单之外的地方。
