1. 从散落函数到可配置接入LangGraph Agent 工具改造的真实痛点如果你正在用 LangGraph 搭 Agent大概率经历过这个阶段一开始只有三五个工具tools.py里写几个函数TOOLS列表一挂路由 prompt 里列一下工具名跑得挺顺。等到工具涨到十几个、还要频繁增删的时候问题就来了——每加一个工具你得改函数、改注册列表、改路由 prompt、改normalize_decision的白名单四处联动漏一处就静默降级到llm排查半天。这篇讲的就是把这种「手写 Tool Calling」的接入方式升级成 MCPModel Context Protocol协议接入。核心变化是工具不再是你代码里散落的函数而是外部服务按统一协议暴露的能力你的 Agent 只需要按配置去连、去发现、去调用。工具增删从「改四处代码」变成「改一段配置」。适合谁看已经在用 LangGraph 或类似框架写 Agent、工具数量开始膨胀、被路由和注册逻辑反复折磨的开发者。读完你能拿到一份可复制的config.toml骨架、MCP server 注册片段以及一次能跑通的调用验证步骤。先说清楚 MCP 和传统 Tool Calling 的区别用个类比传统 Tool Calling 像你自己焊电线接插座每个电器工具都得手动接线MCP 像标准插座面板电器按标准插头插上去就能用面板本身不用管电器内部怎么工作。落到工程上传统方式你要维护「函数实现 工具描述 路由规则 参数校验」四份东西MCP 方式下工具的描述和参数 schema 由服务端通过协议自描述你这边主要是配置连接和做结果适配。我这次改造的场景是本地开发的一个论文阅读 Agent原本有rag、calculator、time、llm四个本地工具现在要加一个联网搜索能力。按老办法得再写一个函数、注册、改路由、改白名单。按 MCP 方式搜索服务由外部提供我只需要在配置里注册这个 MCP server然后处理一下返回结果的解析。下面把整个过程拆开讲。2. TaoToken 统一 Key 前置config.toml 骨架与 MCP server 注册在动 LangGraph 代码之前先把「Key 和模型入口」这件事收敛掉。多工具 Agent 最烦的一点是每个工具、每个模型调用可能用不同的 Key散落在.env、代码常量、环境变量里换一个环境就要重新对一遍。我的做法是用 TaoToken 做统一入口所有模型调用走同一个 Base URL 和同一个 Key配置集中在一个config.toml里。TaoToken 在这里的角色是「统一的模型与能力接入层」你拿一个 Key就能访问它支持的模型对话、编码等能力Agent 里的 LLM 调用、以及需要模型参与的环节都指向同一个入口。官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 这个不加 UTM。Key 在控制台创建地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 创建完在 API Keys 页面能看到页面是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。先给一份config.toml骨架路径放在项目根目录和app/同级# config.toml [llm] base_url https://taotoken.net/api api_key sk-你的TaoTokenKey model claude-sonnet-4-20250514 timeout 60 [mcp.zhipu_search] transport http url https://open.bigmodel.cn/api/mcp/web_search_prime/mcp auth_header Authorization auth_prefix Bearer api_key_env ZHIPU_API_KEY [agent] valid_tools [rag, calculator, time, web_search, llm] web_search_keywords [latest, recent, current, today, news, web, online, 最新, 最近, 当前, 联网] local_doc_keywords [paper1, paper2, this paper, pdf, document, 论文, 文档]这里有几个设计点值得说。第一[llm]段把 Base URL、Key、Model ID 三件套集中管理代码里读配置而不是硬编码换模型只改这一处。第二[mcp.zhipu_search]段描述了一个 MCP server 的连接方式transport是httpurl是服务端点认证走 header。第三[agent]段把工具白名单和路由关键词也配置化这样加工具时改配置就行不用翻代码。读取配置的代码可以这样写放在app/config.pyimport os import tomllib from pathlib import Path CONFIG_PATH Path(__file__).resolve().parent.parent / config.toml with open(CONFIG_PATH, rb) as f: _cfg tomllib.load(f) LLM_BASE_URL _cfg[llm][base_url] LLM_API_KEY _cfg[llm][api_key] CHAT_MODEL _cfg[llm][model] LLM_TIMEOUT _cfg[llm].get(timeout, 60) VALID_TOOLS set(_cfg[agent][valid_tools]) WEB_KEYWORDS _cfg[agent][web_search_keywords] LOCAL_DOC_KEYWORDS _cfg[agent][local_doc_keywords] MCP_SERVERS _cfg.get(mcp, {})注意tomllib是 Python 3.11 起内置的如果你用 3.10 及以下换成tomli并pip install tomli。这样配置和代码就解耦了VALID_TOOLS直接从配置读加工具时改config.toml的valid_tools数组即可。关于 MCP server 的注册langchain-mcp-adapters提供了MultiServerMCPClient它接受一个字典key 是 server 名value 是连接参数。我把这个字典也从配置构造出来而不是写死在代码里def build_mcp_client_config(): servers {} for name, cfg in MCP_SERVERS.items(): entry { transport: cfg[transport], url: cfg[url], } api_key os.getenv(cfg[api_key_env]) if api_key: entry[headers] { cfg[auth_header]: f{cfg[auth_prefix]}{api_key} } servers[name] entry return servers这样config.toml里加一个[mcp.xxx]段代码就自动多一个 server不用改 Python。这就是「工具接入从散落函数收敛为可配置文件」的核心动作。装依赖pip install langchain-mcp-adapters tomli如果你用的是 Claude Code 这类工具做辅助开发它的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面 Base URL、Key、Model ID 的填法和上面config.toml一致可以对照着配。3. 可复制配置MCP 工具封装与 LangGraph 注册片段配置骨架有了接下来把 MCP 工具真正接进 LangGraph。这一步分三块封装 MCP 调用、注册成工具、接入 Agent 流程。每块都给可复制的代码。先看 MCP 调用封装新建app/mcp_tools.py。这里的关键是MultiServerMCPClient拿到工具列表后按名字找到目标工具再ainvoke调用。因为 LangGraph 的节点可能是同步的而 MCP 客户端是异步的所以用asyncio.run包一层import asyncio import json import os from dotenv import load_dotenv from langchain_mcp_adapters.client import MultiServerMCPClient from app.config import build_mcp_client_config from app.logger_config import setup_logger load_dotenv() logger setup_logger() async def _call_mcp_web_search( query: str, recency: str noLimit, content_size: str medium, location: str us, ): client MultiServerMCPClient(build_mcp_client_config()) tools await client.get_tools() search_tool next( (tool for tool in tools if tool.name web_search_prime), None ) if search_tool is None: raise RuntimeError(web_search_prime not found in MCP tools) result await search_tool.ainvoke( { search_query: query, search_recency_filter: recency, content_size: content_size, location: location, } ) return result注意build_mcp_client_config()是从上一节的配置构造的server 名、URL、认证都来自config.toml。这样换 MCP 服务只改配置。MCP 返回的结果结构有时候是「字符串里套 JSON」需要做最多两次json.loads解析这个坑我踩过def _parse_mcp_search_result(raw_result) - list[dict]: if not raw_result: return [] if isinstance(raw_result, list) and len(raw_result) 0: first_item raw_result[0] if isinstance(first_item, dict): raw_text first_item.get(text, ) else: raw_text getattr(first_item, text, ) else: raw_text str(raw_result) data raw_text for _ in range(2): if isinstance(data, str): try: data json.loads(data) except Exception: break if isinstance(data, list): return data return []然后封装成 LangGraph 能用的工具函数把结果整理成带序号的文本方便模型引用def web_search_tool(query: str) - str: logger.info(f[web_search_tool] query: {query}) raw_result asyncio.run( _call_mcp_web_search( queryquery, recencyoneMonth, content_sizemedium, locationus, ) ) items _parse_mcp_search_result(raw_result) if not items: logger.warning([web_search_tool] parsed result is empty) return str(raw_result) lines [] for idx, item in enumerate(items[:5], start1): title item.get(title, No title) link item.get(link, ) content item.get(content, ) lines.append(f[{idx}] {title}\n{content}\n{link}) logger.info([web_search_tool] search finished successfully) return \n\n.join(lines)接着在app/tools.py里注册。这里和传统 Tool Calling 的区别就体现出来了本地工具rag、calculator、time、llm还是本地函数但web_search背后是 MCP 服务注册时只需要把它当成一个普通工具挂进TOOLS列表不需要在注册环节做额外接入from datetime import datetime from app.rag_system import RAGSystem from app.llm_utils import client from app.config import CHAT_MODEL from app.mcp_tools import web_search_tool def rag_tool(query, rag: RAGSystem, chat_historyNone): return rag.ask(query, chat_historychat_history) def calculator_tool(expression): try: return str(eval(expression)) except Exception: return Invalid expression def time_tool(_): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def llm_tool(query, chat_historyNone): messages [{role: system, content: You are a helpful assistant.}] if chat_history: messages.extend(chat_history) messages.append({role: user, content: query}) response client.chat.completions.create( modelCHAT_MODEL, messagesmessages, ) return response.choices[0].message.content TOOLS [ {name: rag, description: Use for paper/document questions, func: rag_tool}, {name: calculator, description: Use for math calculations, func: calculator_tool}, {name: time, description: Get current time, func: time_tool}, { name: web_search, description: Use for external web search when local documents are not enough or when real-time web information is needed, func: web_search_tool, }, {name: llm, description: Use for general questions, func: llm_tool}, ]注意web_search的description写得比较明确强调「本地文档不够时」和「需要实时信息时」这是给路由模型看的判断依据。工具描述的质量直接影响路由准确率别偷懒。最后是接入 LangGraph 流程让normalize_decision接受web_search。原来的白名单是硬编码的现在从配置读from app.config import VALID_TOOLS def normalize_decision(decision: dict, query: str, valid_tool_names: set[str]) - dict: if not isinstance(decision, dict): return {tool: llm, input: query} tool str(decision.get(tool, )).strip().lower() tool_input str(decision.get(input, )).strip() if tool not in valid_tool_names: return {tool: llm, input: query} if tool in {rag, llm, time, web_search}: return {tool: tool, input: query} if tool calculator: if not tool_input: return {tool: calculator, input: query} return {tool: calculator, input: tool_input} return {tool: llm, input: query}valid_tool_names从VALID_TOOLS传入加工具时改config.toml的valid_tools数组这里自动生效。这就是配置化的好处。4. 验证请求一次可复现的 MCP 调用与路由兜底配置和代码都就位了现在验证。验证分两层先单独验证 MCP 调用能通再验证 LangGraph 路由能正确选中web_search。第一层单独跑 MCP 调用。写个临时脚本test_mcp.pyfrom app.mcp_tools import web_search_tool if __name__ __main__: result web_search_tool(LangGraph latest features 2025) print(result[:800])运行前确保.env里有ZHIPU_API_KEY然后python test_mcp.py预期输出是带[1]、[2]序号的搜索结果每条有标题、内容摘要、链接。如果输出是空或者报web_search_prime not found说明 MCP server 连接或工具名有问题去第 5 节排查。第二层验证路由。路由 prompt 要明确告诉模型什么时候选web_search。我调整后的 prompt 片段prompt f You are a tool router. Your job is ONLY to choose the best tool and prepare its input. Do NOT answer the users question. Do NOT rewrite the users question into an answer. Return JSON only. Available tools: {tool_desc} Tool selection guidance: - Use rag for questions about the loaded local papers/documents. - Use calculator for clear math calculations. - Use time for current time questions. - Use web_search for questions that explicitly need web information, latest information, recent updates, current events, online search, or information likely not contained in the local PDFs. - Use llm for general questions that do not need document retrieval, calculation, time, or web search. Rules: 1. You must return exactly one JSON object. 2. JSON format: {{tool: ..., input: ...}} 3. tool must be one of: rag, calculator, time, web_search, llm 4. For rag, llm, time, and web_search: input should stay the same as the users original question. 5. For calculator: input should be the math expression only if you can extract it. 6. Do not include markdown, explanations, or code fences. User question: {query} 光靠模型软路由不够稳因为模型判断有不确定性。加一个刚性兜底规则用关键词匹配强制走web_searchfrom app.config import WEB_KEYWORDS, LOCAL_DOC_KEYWORDS def maybe_force_web_search(query: str, decision: dict) - dict: q query.lower() has_web_signal any(k in q for k in WEB_KEYWORDS) looks_like_local_doc any(k in q for k in LOCAL_DOC_KEYWORDS) if has_web_signal and not looks_like_local_doc: return {tool: web_search, input: query} return decision注意这里有个细节如果问题里同时出现「论文」和「最新」比如「这篇论文的最新进展」looks_like_local_doc为真就不强制走 web交给模型判断。这个优先级设计避免误伤本地文档查询。在choose_tool_node里把两步串起来raw_decision json.loads(cleaned) decision normalize_decision(raw_decision, query, valid_tool_names) decision maybe_force_web_search(query, decision) logger.info(f[choose_tool_node] raw decision: {raw_decision}) logger.info(f[choose_tool_node] normalized decision: {decision}) return {decision: decision}验证路由跑几个测试 queryfrom app.graph import build_graph graph build_graph() for q in [最近 LangGraph 有什么更新, paper1 讲了什么, 3 加 5 等于几]: result graph.invoke({query: q}) print(q, -, result[decision])预期结果第一个走web_search第二个走rag第三个走calculator。如果第一个走了llm检查WEB_KEYWORDS里有没有「最近」以及maybe_force_web_search有没有被调用。实测下来加了刚性兜底后含「最新」「最近」「联网」这类词的 query 命中web_search的准确率明显提升不再依赖模型每次判断都对。这就是软硬结合的价值模型负责语义理解关键词负责兜底。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth改造过程中我遇到几个典型报错逐个说清楚现象、原因、解法。报错一401 Unauthorized。现象是调用 MCP 或 LLM 时返回 401。原因通常是 Key 没读到或格式不对。排查顺序先确认.env里ZHIPU_API_KEY和config.toml里api_key都填了再确认build_mcp_client_config里auth_prefix是Bearer注意有个空格最后确认 Key 没有多余引号或换行。如果是 TaoToken 的 Key 报 401去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 重新确认 Key 状态。报错二local proxy failed。现象是连接 MCP server 时报代理相关错误。这个多半是环境变量里有HTTP_PROXY、HTTPS_PROXY之类的残留配置导致请求走了不该走的路径。解法是检查环境变量把不需要的代理配置清掉或者在代码里显式指定不走代理。注意这里说的是清理本地环境变量不是让你去配什么网络工具。报错三reading choices of undefined。现象是llm_tool里response.choices[0]报错。原因是模型调用返回结构异常可能是 Base URL 或 Model ID 配错请求没真正到达模型。排查确认config.toml里base_url是https://taotoken.net/apimodel是有效 Model ID。可以先用模型对话页面单独验证模型能不能通地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 能通说明 Key 和模型没问题问题在代码调用层。报错四OAuth 相关错误。如果你用 Claude Code 或类似工具做辅助开发接入时可能遇到 OAuth 报错。这类工具接入需要填全三件套Base URL、Key、Model ID。Base URL 填https://taotoken.net/apiKey 填控制台创建的 KeyModel ID 填你要用的模型。三件套缺一个或者填错都会报 OAuth 或认证失败。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 对照着填。报错五web_search_prime not found。现象是 MCP 工具列表里找不到目标工具。原因是 server 名或工具名对不上。解法先打印tools列表看实际有哪些工具名再改next((tool for tool in tools if tool.name ...), None)里的名字。不同 MCP 服务的工具命名不一样别照抄。报错六asyncio.run 报 event loop 已运行。现象是在已有事件循环的环境里调asyncio.run报错。原因是web_search_tool是同步函数内部用asyncio.run如果外层已经在异步上下文里就会冲突。解法要么把工具改成异步要么用nest_asyncio打补丁要么把 MCP 调用放到独立线程。本地开发场景下最简单的是确保调用链是同步的。排查这类问题的通用思路先隔离验证单独跑 MCP 调用、单独跑模型调用确认哪一层出问题再针对性修。别一上来就改一堆代码那样只会引入新问题。6. 语义一致 CTA把工具接入收敛成配置然后继续往下走回到开头那个痛点工具一多注册、路由、白名单四处联动改一处漏一处。这篇给的解法是把这些收敛到config.tomlvalid_tools管白名单[mcp.xxx]管 MCP server 注册web_search_keywords管兜底规则。加工具时改配置代码不动。这就是从 Tool Calling 到 MCP 的工程升级——不是换个调用方式而是把「工具接入」这件事从代码逻辑里抽出来变成可配置、可复用、可增删的模块。如果你还在用散落函数的方式接工具建议先做这一步收敛哪怕暂时不接 MCP把白名单和关键词配置化也能省不少事。等你需要接外部能力时MCP 的配置化接入就是顺理成章的下一步。接下来可以往两个方向走。一是把更多外部能力接进来比如代码执行、数据库查询都按[mcp.xxx]的格式加配置。二是把 Agent 跑得更稳比如加工具调用失败重试、结果缓存。如果你要长期跑编码类 Agent可以看看 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 它针对长时间编码场景做了优化。模型对话验证在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。最后留个实用技巧MCP 工具的description一定要写清楚「什么时候用」和「什么时候不用」这是路由准确率的关键。我试过把web_search的描述从「搜索网络」改成「本地文档不够或需要实时信息时使用」路由命中率提升很明显。工具描述是给模型看的接口文档值得多花几分钟打磨。