客户端接入实战：在 LangChain 中集成 MCP 工具调用与 TaoToken 统一 Key 配置

1. 为什么要在 LangChain 里接 MCP而不是继续手写工具函数如果你已经用 LangChain 写过几个 Agent大概率经历过这样的阶段一开始把工具函数用tool装饰器包一层直接塞进tools列表里跑得挺顺。等到工具数量涨到十几个、还要跨团队复用的时候问题就来了——每个项目都要复制一遍工具定义参数 schema 各写各的改一个字段要同步好几个仓库。MCPModel Context Protocol解决的正是这件事。它把「工具提供方」和「工具调用方」拆开服务端只负责注册工具、声明参数 schema、执行逻辑客户端负责发现工具、构造调用、拿回结果。LangChain 作为客户端接入 MCP 之后Agent 看到的工具列表是动态从服务端拉取的你不需要在 LangChain 代码里硬编码任何工具签名。这篇要跑通的闭环是本地起一个 MCP 服务端stdio 模式LangChain 客户端通过 MCP 协议发现工具把每个远程工具动态包装成BaseTool再交给 Agent 调用。同时所有模型请求统一走 TaoToken 的 API 通道用一个 Key 管理多模型调用避免在 LangChain 里散落一堆厂商 Key。适合谁看已经能跑通 LangChain Agent、想把手写工具迁移到 MCP 的开发者或者正在做多模型统一接入、希望 Key 管理收敛到一处的团队。下面所有配置和代码都可以直接复制到本地跑。2. TaoToken 前置统一 Key 与 API 通道准备在写 LangChain 适配器之前先把模型调用这一层收敛掉。LangChain 里如果同时用 OpenAI、Anthropic、通义等模型每个都要配OPENAI_API_KEY、ANTHROPIC_API_KEY环境变量一多就容易乱。TaoToken 提供的是 OpenAI 兼容的 API 通道LangChain 的ChatOpenAI可以直接指过去一个 Key 覆盖多个模型。你需要准备的东西只有两样一个 API Key一个 base_url。Key 在控制台的 API Keys 页面创建建议按项目建独立 Key方便后续按 Key 统计用量和吊销。控制台入口创建/管理 Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsoleAPI Keys 直达https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi-keys接入文档参数、模型名对照https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdocbase_url 用https://taotoken.net/api注意这个地址后面不加任何路径后缀LangChain 的 OpenAI 兼容层会自动拼/chat/completions。模型名按文档里给的写比如gpt-4o、claude-3-5-sonnet这类具体以文档为准。注意Key 只放在环境变量或本地配置文件里不要提交到 Git。下面示例统一用TAOTOKEN_API_KEY这个环境变量名。3. 可复制配置config.toml 与 settings.json 骨架MCP 客户端和服务端的连接参数、LangChain 的模型参数建议分两个文件管理config.toml放 MCP 服务端启动命令和超时settings.json放模型与 Key 相关配置。这样换环境时只改配置不动代码。先看config.toml它描述的是「怎么启动 MCP 服务端」以及「客户端连接行为」# config.toml [mcp] transport stdio command python args [-m, db_mcp_server] timeout_seconds 30 retry_attempts 3 retry_backoff 1.5 [mcp.env] DB_PATH ./data/app.db LOG_LEVEL INFO [langchain] model gpt-4o temperature 0.2 max_tokens 1024transport目前用stdio最省事服务端作为子进程启动客户端通过标准输入输出通信。timeout_seconds和重试参数是给工具调用兜底的后面排障会用到。再看settings.json它管的是模型通道和 Key 的读取方式{ llm: { provider: openai-compatible, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: gpt-4o, fallback_models: [claude-3-5-sonnet, gpt-4o-mini] }, mcp: { config_path: ./config.toml, tool_prefix: mcp_, enable_stream: true }, logging: { level: INFO, trace_tool_calls: true } }api_key_env写的是环境变量名而不是 Key 本身代码里读这个字段再去取环境变量。tool_prefix给所有 MCP 工具加统一前缀避免和本地工具重名。fallback_models是备用模型列表主模型超时或限流时可以切换。设置环境变量Linux/macOSexport TAOTOKEN_API_KEY你的KeyWindows PowerShell$env:TAOTOKEN_API_KEY你的Key4. 客户端接入动态发现工具并包装成 LangChain BaseTool这一节是核心。思路分三步连上 MCP 服务端、拉取工具元数据、把每个工具动态生成一个BaseTool子类实例。先装依赖。MCP 官方 Python SDK 迭代较快建议锁版本pip install mcp0.1.0a4 langchain langchain-openai pydantic tomli读取配置和建立连接的代码import asyncio import os import tomli from mcp import Client, StdioServerParameters def load_mcp_params(config_path: str) - StdioServerParameters: with open(config_path, rb) as f: cfg tomli.load(f) mcp_cfg cfg[mcp] return StdioServerParameters( commandmcp_cfg[command], argsmcp_cfg[args], env{**os.environ, **mcp_cfg.get(env, {})}, )接下来是适配器。关键点在于MCP 返回的inputSchema是 JSON Schema要转成 Pydantic 模型给 LangChain 用_arun里调用client.call_tool把结果从 JSON-RPC 响应里取出来。from typing import Any, Type from pydantic import BaseModel, Field, create_model from langchain.tools import BaseTool from mcp import Client JSON_TYPE_MAP { string: str, integer: int, number: float, boolean: bool, array: list, object: dict, } def build_args_schema(tool_name: str, input_schema: dict) - Type[BaseModel]: fields: dict[str, Any] {} props input_schema.get(properties, {}) required set(input_schema.get(required, [])) for name, spec in props.items(): py_type JSON_TYPE_MAP.get(spec.get(type, string), str) default ... if name in required else None fields[name] ( py_type, Field(defaultdefault, descriptionspec.get(description, )), ) return create_model(f{tool_name}_Args, **fields) class MCPToolAdapter(BaseTool): name: str description: str args_schema: Type[BaseModel] client: Client def _run(self, **kwargs) - str: raise NotImplementedError(请使用异步执行器 arun) async def _arun(self, **kwargs) - str: resp await self.client.call_tool(self.name, argumentskwargs) contents resp.get(content, []) texts [c.get(text, ) for c in contents if c.get(type) text] return \n.join(texts) if texts else str(resp)注意_run直接抛异常强制走异步。原因在排障一节会讲——同步方法里asyncio.run在已有事件循环的环境下会炸。工具工厂负责把元数据批量转成适配器实例async def create_mcp_tools(client: Client, prefix: str mcp_) - list[MCPToolAdapter]: meta await client.list_tools() tools [] for item in meta[tools]: schema build_args_schema(item[name], item.get(inputSchema, {})) tools.append( MCPToolAdapter( namef{prefix}{item[name]}, descriptionitem.get(description, ), args_schemaschema, clientclient, ) ) return tools5. 验证请求一次可复制的工具调用闭环现在把模型、工具、Agent 串起来。模型走 TaoToken 通道用ChatOpenAI指定base_url和api_keyimport asyncio import json import os from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from mcp import Client async def main(): with open(settings.json, r, encodingutf-8) as f: settings json.load(f) llm ChatOpenAI( modelsettings[llm][default_model], base_urlsettings[llm][base_url], api_keyos.environ[settings[llm][api_key_env]], temperature0.2, ) params load_mcp_params(settings[mcp][config_path]) async with Client(params) as client: tools await create_mcp_tools(client, settings[mcp][tool_prefix]) print(发现工具:, [t.name for t in tools]) agent create_react_agent(llm, tools, prompt你是一个数据库助手可以查询用户信息。) executor AgentExecutor(agentagent, toolstools, verboseTrue) result await executor.ainvoke({input: 查询年龄大于30岁的用户有哪些}) print(最终结果:, result[output]) if __name__ __main__: asyncio.run(main())跑起来后终端会先打印发现到的工具列表然后 Agent 进入推理循环。正常输出大致是这样发现工具: [mcp_query_db] Entering new AgentExecutor chain... Action: mcp_query_db Action Input: {sql: SELECT name FROM users WHERE age 30} Observation: [{name: 张三}, {name: 李四}] Final Answer: 年龄大于30岁的用户有张三和李四。看到Final Answer就说明闭环通了LangChain 客户端通过 MCP 协议发现了远程工具动态包装后交给 AgentAgent 调用工具拿到结果模型再基于结果生成回答。整个过程 LangChain 代码里没有出现任何query_db的硬编码签名。如果你想单独验证模型通道是否正常可以先用模型对话页面发一条消息确认 Key 和模型名没问题https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentchat6. 本篇常见错排查Connection refused / 服务端进程退出先确认config.toml里的command和args能在终端直接跑起来。手动执行python -m db_mcp_server看是否报模块找不到或数据库路径错误。stdio 模式下服务端是子进程它的 stderr 不会自动显示建议在服务端代码里把日志写到文件方便定位。Tool not found: mcp_xxx两种可能一是list_tools返回的元数据里没有这个工具检查服务端注册逻辑二是前缀不一致Agent 看到的工具名带了mcp_前缀但你在 prompt 或调用里写的是原始名。统一用create_mcp_tools返回的name字段。JSON-RPC error: -32603服务端内部错误通常是工具执行时抛了异常。比如 SQL 语法错误、数据库连接失败。在服务端的call_tool处理函数里加 try/except把异常信息通过isError字段返回客户端就能看到具体原因。asyncio.run() cannot be called from a running event loop这是同步_run里调asyncio.run的经典问题。LangChain 的AgentExecutor.ainvoke本身就在事件循环里再嵌套一个asyncio.run必然报错。解决办法就是本篇的做法_run直接抛异常统一走_arun用ainvoke而不是invoke。模型返回 401 或 model not found检查TAOTOKEN_API_KEY是否设置成功base_url是否写成https://taotoken.net/api不要带/v1或/chat/completions。模型名以接入文档为准写错模型名会返回 not found。如果主模型限流可以在ChatOpenAI外层包一层 fallback或者临时切到fallback_models里的模型。工具调用超时config.toml里的timeout_seconds默认 30 秒长查询可能不够。调大超时的同时建议在服务端对耗时操作做分页或异步任务化避免客户端一直等。重试逻辑用retry_attempts和retry_backoff控制指数退避能缓解偶发网络抖动。7. 长期编码与 Agent 场景的 Key 管理建议如果你打算把 MCP 工具调用用在日常编码或长期跑的 Agent 上Key 管理值得单独规划。我的做法是按用途拆 Key一个用于本地开发调试一个用于 CI 或定时任务一个用于生产 Agent。这样某个 Key 出问题或需要轮换时影响面可控。TaoToken 的 Coding Plan 适合需要长期、稳定调用多模型的编码场景配合 LangChain 的 Agent 循环可以把工具调用和模型推理的用量分开观察https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding-plan接入文档里有完整的参数说明和模型名对照遇到配置对不上时优先查这里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc最后提醒一个工程细节MCP 客户端连接不要每次工具调用都重建。本篇示例用async with Client(params)包住整个 Agent 执行过程连接在会话内复用。如果工具调用频率很高可以在外层维护一个长连接配合心跳检测避免频繁握手带来的延迟。