1. 从一堆散装工具函数说起:LangChain 接 MCP 到底解决什么问题
如果你用 LangChain 写过带工具的 Agent,大概率经历过这个阶段:每接一个外部能力,就要手写一个@tool函数,参数 schema 自己定,错误处理自己兜,工具一多,tools=[...]列表长得像流水账。更麻烦的是,同一个数据库查询工具,你在 LangChain 里写一遍,换到别的框架又得重写一遍,工具和框架被死死绑在一起。
MCP(Model Context Protocol)想干的事,就是把这层绑定解开。它定义了一套标准协议,让工具提供方只需要实现一个 MCP Server,任何支持 MCP 的客户端(LangChain、各类 Agent 运行时)都能直接挂载使用。对 LangChain 应用来说,MCP 服务就是「可插拔的工具箱」:进程启动时握手,动态拉取工具列表,转成 LangChain 的 Tool 对象,Agent 照常调用,完全不用关心工具内部是读文件还是查数据库。
这篇面向的是已经能跑通基础 LangChain Agent、想进一步做「多 MCP 服务注册 + 路由 + 统一配置」的开发者。我会给出一份可复制的config.toml与settings.json骨架,把多个 MCP Server 的启动参数、工具命名空间、路由规则集中管理,再配合 TaoToken 的模型接入配置,最后给出启动后验证 MCP 连通性的具体动作。整套结构的目标是:加一个新工具服务,只改配置文件,不动业务代码。
2. TaoToken 前置:把模型入口和 Key 先理顺
MCP 负责工具侧,模型侧我建议单独抽出来。原因很简单:Agent 跑起来之后,工具调用会频繁触发多轮模型请求,如果模型入口散落在代码里,换模型、调并发、排查限流都会很痛苦。我习惯把模型统一走 TaoToken 的 API 入口,Key 和 base_url 集中在一处配置。
先拿到访问凭证。打开控制台创建 API Key,建议按项目建独立的 Key,方便后续按项目看用量:
https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=console创建完 Key 之后,模型调用的 base_url 统一填https://taotoken.net/api,这个地址不加任何查询参数,直接作为 OpenAI 兼容的 endpoint 使用。LangChain 里用ChatOpenAI时把base_url和api_key指过去就行,后面配置文件里我会把它写成环境变量引用,避免硬编码。
如果你还没确定用哪个模型跑 Agent,可以先去模型对话页面手动试几轮工具调用类的 prompt,确认模型对 function calling 的响应格式稳定,再写进配置:
https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=model-chat需要长期跑编码类 Agent、或者工具调用轮次特别多的场景,可以看下 Coding Plan,它在高频调用下的额度策略比按次计费更划算:
https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-planKey 的管理入口在这里,后续如果要做多环境(dev/staging)隔离,可以在这里建多个 Key:
https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=api-keys接入细节和参数说明以官方文档为准:
https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc3. 可复制配置:config.toml 管 MCP 服务,settings.json 管运行时
多 MCP 服务最容易乱的地方是「启动参数散落在代码里」。我的做法是分两层:config.toml描述有哪些 MCP Server、怎么启动、工具挂到哪个命名空间;settings.json描述运行时行为,比如模型参数、路由策略、超时。这样加服务只动 toml,调行为只动 json。
先看config.toml。每个[[mcp_servers]]块对应一个 MCP 服务,namespace用来给工具名加前缀,避免不同服务出现同名工具时冲突:
# config.toml [app] name = "langchain-mcp-demo" log_level = "INFO" [model] # 模型统一走 TaoToken 的 OpenAI 兼容入口 base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" default_model = "gpt-4o" temperature = 0.0 request_timeout = 60 # 第一个 MCP 服务:文件系统 [[mcp_servers]] name = "filesystem" namespace = "fs" transport = "stdio" command = "npx" args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-demo"] enabled = true tool_allowlist = ["read_file", "write_file", "list_directory"] # 第二个 MCP 服务:SQLite [[mcp_servers]] name = "sqlite" namespace = "db" transport = "stdio" command = "npx" args = ["-y", "@modelcontextprotocol/server-sqlite", "/tmp/mcp-demo/demo.db"] enabled = true tool_allowlist = ["read_query", "list_tables"] # 第三个 MCP 服务:预留的远程服务,先禁用 [[mcp_servers]] name = "remote-tools" namespace = "remote" transport = "sse" url = "http://127.0.0.1:8765/sse" enabled = false几个关键点说明一下。transport目前主流是stdio和sse两种,本地进程用 stdio,独立部署的服务用 sse。tool_allowlist是安全边界,只放行你确认要暴露给 Agent 的工具,别图省事全开。namespace会在加载时拼成fs.read_file这种形式,Agent 看到的工具名带前缀,路由时按前缀分发。
再看settings.json,它管的是运行时策略:
{ "agent": { "max_iterations": 8, "verbose": true, "handle_parsing_errors": true }, "routing": { "strategy": "namespace_prefix", "rules": [ { "prefix": "fs.", "server": "filesystem" }, { "prefix": "db.", "server": "sqlite" }, { "prefix": "remote.", "server": "remote-tools" } ] }, "mcp": { "connect_timeout_seconds": 15, "tool_refresh_interval_seconds": 300, "fail_fast": false }, "observability": { "log_tool_calls": true, "log_model_requests": false } }fail_fast设成 false 是有意的:某个 MCP 服务启动失败时,应用不应该整体崩掉,而是跳过它继续加载其他服务,日志里标记出来即可。tool_refresh_interval_seconds用于定期重新拉取工具列表,适合工具会动态变化的远程服务。
加载配置的代码骨架大概长这样,用tomllib读 toml,用json读 settings,然后按 namespace 组装工具:
import json import tomllib from pathlib import Path from langchain_mcp_adapters.client import MultiServerMCPClient def load_config(config_path: str = "config.toml"): with open(config_path, "rb") as f: return tomllib.load(f) def load_settings(settings_path: str = "settings.json"): return json.loads(Path(settings_path).read_text(encoding="utf-8")) def build_mcp_client(cfg: dict) -> MultiServerMCPClient: servers = {} for s in cfg.get("mcp_servers", []): if not s.get("enabled", True): continue if s["transport"] == "stdio": servers[s["name"]] = { "command": s["command"], "args": s["args"], "transport": "stdio", } elif s["transport"] == "sse": servers[s["name"]] = { "url": s["url"], "transport": "sse", } return MultiServerMCPClient(servers)这段代码里MultiServerMCPClient负责同时管理多个 MCP 连接,工具加载时会把每个服务返回的工具合并成一个列表。命名空间前缀可以在拿到工具后手动重命名,也可以在路由层做映射,我倾向后者,保持工具原始名不变,路由时按server字段分发。
4. 启动后验证 MCP 连通性:三个具体动作
配置写完不代表能跑通,MCP 是进程间通信,握手失败、工具列表为空、schema 不兼容都很常见。我一般按下面三步验证,每步都有明确的成功标志。
第一步,单独验证 MCP Server 能启动。不要一上来就跑整个 Agent,先用最原始的方式确认服务进程本身没问题:
npx -y @modelcontextprotocol/server-filesystem /tmp/mcp-demo如果进程能起来并停在等待输入的状态,说明命令和参数没问题。stdio 模式下它不会打印太多东西,这是正常的,按 Ctrl+C 退出即可。如果报模块找不到,检查 Node 版本和 npx 缓存。
第二步,用 Python 脚本单独拉一次工具列表,确认握手和工具发现成功:
import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient async def check_tools(): client = MultiServerMCPClient({ "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-demo"], "transport": "stdio", } }) tools = await client.get_tools() for t in tools: print(f"tool={t.name} | desc={t.description[:60]}") print(f"total_tools={len(tools)}") asyncio.run(check_tools())成功标志是打印出read_file、write_file、list_directory这几个工具名,且total_tools大于 0。如果列表为空,多半是 args 路径不对或者服务启动即退出,把npx命令手动跑一遍对比。
第三步,跑一个最小 Agent 任务,验证工具调用链路完整。这一步会真正触发模型请求和 MCP 工具执行:
import asyncio import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_mcp_adapters.client import MultiServerMCPClient async def run_agent(): client = MultiServerMCPClient({ "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/mcp-demo"], "transport": "stdio", } }) tools = await client.get_tools() llm = ChatOpenAI( model="gpt-4o", temperature=0, base_url="https://taotoken.net/api", api_key=os.environ["TAOTOKEN_API_KEY"], ) prompt = ChatPromptTemplate.from_messages([ ("system", "你可以调用工具读写文件,请根据用户要求选择合适工具。"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_tool_calling_agent(llm, tools, prompt) executor = AgentExecutor(agent=agent, tools=tools, verbose=True) result = await executor.ainvoke({ "input": "在 /tmp/mcp-demo 下创建 hello.txt,写入 Hello MCP" }) print(result["output"]) asyncio.run(run_agent())成功标志有两个:verbose 日志里能看到工具调用记录,且/tmp/mcp-demo/hello.txt文件真实存在、内容正确。到这一步,说明模型入口、MCP 握手、工具路由、文件写入整条链路都通了。
5. 本篇常见错排查
报错一:MCP server failed to start: command not found
这是最常见的一类,本质是 stdio 模式下子进程启动失败。先确认command在 PATH 里,npx这类命令在部分环境下需要写绝对路径。其次检查args里的路径是否存在,MCP Server 对不存在的目录经常直接退出而不报错。排查方法就是把command + args拼成一条 shell 命令手动执行,看真实报错。
报错二:工具列表为空,但进程能启动
多半是握手超时或协议版本不匹配。把connect_timeout_seconds调大到 30 再试。如果用的是较老的 MCP Server 实现,可能返回的工具 schema 和当前适配器不兼容,此时看日志里有没有 schema 解析警告。另一个常见原因是tool_allowlist写错了工具名,导致全部被过滤掉,先临时去掉 allowlist 验证。
报错三:模型不调用工具,直接编答案
这不是 MCP 的问题,是模型侧的问题。检查两点:一是工具描述是否清晰,MCP Server 返回的 description 如果太模糊,模型不知道何时该用;二是 prompt 里有没有明确引导使用工具。另外确认base_url指向的是https://taotoken.net/api,如果模型入口配错,function calling 的响应格式可能不被正确解析。
报错四:多服务下工具名冲突
两个 MCP Server 都提供read_file时,合并后的工具列表会出现重名,Agent 调用时行为不确定。解决办法就是配置里的namespace前缀,加载后统一重命名,或者用settings.json里的路由规则按 server 分发。我建议在加载阶段就完成重命名,别留到调用时再判断。
报错五:SSE 模式连不上远程服务
先确认远程服务确实在监听,用 curl 打一下/sse端点看有没有响应。SSE 模式对网络环境比 stdio 敏感,本地开发建议先用 stdio 跑通逻辑,再切 SSE 部署。如果远程服务需要鉴权,检查 header 配置是否传对。
6. 把配置骨架用起来:下一步怎么扩展
这套结构的核心价值在于「配置驱动」:新增一个 MCP 服务,只需要在config.toml里加一个[[mcp_servers]]块,在settings.json的路由规则里加一条前缀映射,业务代码一行不用改。工具加载、命名空间、超时、日志这些横切关注点都收敛在配置层。
实际项目里我还会做两件事。一是给每个 MCP 服务加健康检查,启动时并发探测,把不可用的服务标记出来但不阻塞主流程;二是把工具调用日志单独落一份,方便回溯 Agent 到底调了哪些工具、传了什么参数、返回了什么,这对排查「模型为什么没选对工具」特别有用。
模型侧继续走 TaoToken 的统一入口,Key 用环境变量注入,多环境用不同 Key 隔离。工具侧按 MCP 标准协议扩展,本地进程用 stdio,独立服务用 sse,配置里切换 transport 即可。这样一套下来,LangChain 应用的工具生态就是可插拔的,加服务像加配置项一样轻。