用 toolbox-langchain 打通数据库MCP Toolbox Python SDK 的 LangChain/LangGraph 集成实战【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox导读本文聚焦 MCP Toolbox 官方 Python SDK 中的toolbox-langchain包讲解如何在一个正在运行的 MCP Toolbox 服务之上把数据库中定义好的工具Tool / Toolset加载进 LangChain 与 LangGraph 应用构建具备真实数据库操作能力的 Agent。读完本文你将掌握客户端初始化与 MCP 协议版本选择、工具加载、与 LangChain / LangGraph 的三种接入模式ReAct Agent、节点式图、手动调用以及客户端到服务端认证、工具级认证、参数绑定、安全参数Secure Parameters与 OpenTelemetry 可观测性等生产级配置并能在当前仓库中找到对应的服务端实现与完整示例代码作为佐证。Overviewtoolbox-langchain 是什么toolbox-langchain是 MCP Toolbox 官方 Python SDK 中面向 LangChain 生态的适配层。它为 MCP Toolbox 服务提供了一个 Python 接口使你可以在自己的应用中加载并调用由 Toolbox 服务托管的工具——这些工具通常由tools.yaml之类的服务端配置定义本质上是封装了 SQL 语句、数据库操作或其他 API 调用的可执行单元。它在 SDK 栈中的位置与 Python Core SDK 一脉相承toolbox-core提供最底层的ToolboxClient与协议协商能力toolbox-langchain则把加载出的工具整理成符合 LangChain 工具约定的可调用对象可以直接交给bind_tools()、create_react_agent或ToolNode使用。因此本文中的大部分概念传输协议、认证、参数绑定、安全参数、遥测都能在 Core SDK 文档中找到更底层的对应实现。安装与运行环境准备安装 SDK 只需一条命令pip install toolbox-langchain在使用 SDK 之前需要保证 Toolbox 服务已经在本地 5000 端口运行。完整的端到端搭建流程数据库准备、Toolbox 服务安装与配置、Agent 连接可以参阅仓库内的 Toolbox Quickstart 教程。如果你希望直接跑通一个 LangChain 场景仓库中已经提供了可运行的完整示例 quickstart.py该示例演示了用ChatGoogleGenerativeAI也可替换为ChatAnthropic配合create_react_agent完成酒店搜索、预订、取消与改期四轮对话import asyncio from langgraph.prebuilt import create_react_agent # TODO(developer): replace this with another import if needed from langchain_google_genai import ChatGoogleGenerativeAI # from langchain_anthropic import ChatAnthropic from langgraph.checkpoint.memory import MemorySaver from toolbox_langchain import ToolboxClient prompt Youre a helpful hotel assistant. You handle hotel searching, booking and cancellations. When the user searches for a hotel, mention its name, id, location and price tier. Always mention hotel ids while performing any searches. This is very important for any operations. For any bookings or cancellations, please provide the appropriate confirmation. Be sure to update checkin or checkout dates if mentioned by the user. Dont ask for confirmations from the user. queries [ Find hotels in Basel with Basel in its name., Can you book the Hilton Basel for me?, Oh wait, this is too expensive. Please cancel it and book the Hyatt Regency instead., My check in dates would be from April 10, 2024 to April 19, 2024., ] async def main(): # TODO(developer): replace this with another model if needed model ChatGoogleGenerativeAI(modelgemini-2.5-flash) # model ChatAnthropic(modelclaude-3-5-sonnet-20240620) # Load the tools from the Toolbox server async with ToolboxClient(http://127.0.0.1:5000) as client: tools await client.aload_toolset() agent create_react_agent(model, tools, checkpointerMemorySaver()) config {configurable: {thread_id: thread-1}} for query in queries: inputs {messages: [(user, prompt query)]} print(f\n[INPUT] User: {query}) response agent.invoke(inputs, stream_modevalues, configconfig) print(f[OUTPUT] AI: {response[messages][-1].content}) asyncio.run(main())这个示例展示了三个关键点ToolboxClient作为异步上下文管理器使用、aload_toolset()一次性加载整个工具集、以及MemorySaver让 Agent 在多轮对话中保持状态这正是 LangGraph 相比普通 LangChain 调用的核心优势。Quickstart最小可运行示例官方文档给出了一个最精简的入门示例使用 LangGraph 的create_react_agent构建 Agentfrom toolbox_langchain import ToolboxClient from langchain_google_vertexai import ChatVertexAI from langgraph.prebuilt import create_react_agent async with ToolboxClient(http://127.0.0.1:5000) as toolbox: tools toolbox.load_toolset() model ChatVertexAI(modelgemini-3-flash-preview) agent create_react_agent(model, tools) prompt Hows the weather today? for s in agent.stream({messages: [(user, prompt)]}, stream_modevalues): message s[messages][-1] if isinstance(message, tuple): print(message) else: message.pretty_print()要点拆解ToolboxClient(http://127.0.0.1:5000)指向 Toolbox 服务地址async with负责客户端的生命周期管理关闭底层网络会话。toolbox.load_toolset()不带参数时加载服务端配置的全部工具集。create_react_agent(model, tools)是 LangGraph prebuilt 提供的 ReAct 风格 Agent自动完成思考 → 调用工具 → 观察结果 → 继续的循环。agent.stream(..., stream_modevalues)逐轮输出消息pretty_print()负责格式化打印。完整的多轮对话版本同样见 quickstart.py。如果需要在构建 Agent 前先完整掌握服务端搭建请先阅读 Toolbox Quickstart 教程。初始化客户端与传输协议基础初始化导入并初始化客户端指向正在运行的 Toolbox 服务from toolbox_langchain import ToolboxClient # Replace with your Toolbox services URL async with ToolboxClient(http://127.0.0.1:5000) as toolbox:支持的传输协议SDK 支持多种与 Toolbox 服务端通信的传输协议默认使用当前最新稳定版本的Model Context Protocol (MCP)。可以通过初始化时的protocol参数显式选择协议例如需要使用 Toolbox 原生 HTTP 协议或希望把客户端固定到某个 MCP 历史版本时这都非常有用。所有 MCP 传输选项都是基于Model Context Protocol over HTTP实现的。常量说明Protocol.MCP默认默认 MCP 版本的别名当前为2026-07-28。Protocol.MCP_LATEST最新稳定 MCP 版本的别名当前为2026-07-28。Protocol.MCP_DRAFT即将发布的草稿 MCP 版本别名当前为2026-07-28。Protocol.MCP_v20260728MCP 协议版本 2026-07-28。Protocol.MCP_v20251125MCP 协议版本 2025-11-25。Protocol.MCP_v20250618MCP 协议版本 2025-06-18。Protocol.MCP_v20250326MCP 协议版本 2025-03-26。Protocol.MCP_v20241105MCP 协议版本 2024-11-05。从源码结构看这些协议版本并不是虚构的仓库的服务端 MCP 实现中维护了与之一一对应的版本目录例如 internal/server/mcp/v20241105、internal/server/mcp/v20250326、internal/server/mcp/v20250618、internal/server/mcp/v20251125 与 internal/server/mcp/v20260728每个目录都实现了该版本的 JSON-RPC 消息处理印证了协议协商是端到端真实生效的机制。默认协议示例from toolbox_langchain import ToolboxClient from toolbox_core.protocol import Protocol async with ToolboxClient(http://127.0.0.1:5000, protocolProtocol.MCP) as toolbox: # Use client pass固定到 MCP 2025-03-26 版本from toolbox_langchain import ToolboxClient from toolbox_core.protocol import Protocol async with ToolboxClient(http://127.0.0.1:5000, protocolProtocol.MCP_v20250326) as toolbox: # Use client pass需要说明的是Core SDK 文档中补充了两种更进阶的用法Core SDK 传输协议一是传入协议列表做协商回退如protocol[Protocol.MCP_LATEST, Protocol.MCP_v20250618]二是传入仅含单个值的数组以严格固定版本并禁用回退如protocol[Protocol.MCP_DRAFT]。加载工具加载一个工具集Toolset工具集是一组相关工具的集合可以加载其中的全部工具也可以只加载某个指定工具集# Load all tools tools toolbox.load_toolset() # Load a specific toolset tools toolbox.load_toolset(my-toolset)加载单个工具tool toolbox.load_tool(my-tool)加载单个工具能让你对哪些工具对 LLM Agent 可见拥有更细粒度的控制这在安全敏感或工具数量庞大的场景下尤其有用。工具本身的定义方式kind: tool、参数列表、SQL 语句等见 工具配置文档。与 LangChain 集成LangChain 的 Agent 会根据用户输入动态选择并执行工具。将从 Toolbox SDK 加载的工具加入 Agent 的工具包即可from langchain_google_vertexai import ChatVertexAI model ChatVertexAI(modelgemini-3-flash-preview) # Initialize agent with tools agent model.bind_tools(tools) # Run the agent result agent.invoke(Do something with the tools)bind_tools(tools)会把工具 schema 注入模型请求使模型在需要时能够发起工具调用。这里传入的tools就是load_toolset()/load_tool()的返回值。与 LangGraph 集成将 Toolbox SDK 与 LangGraph 集成可以让你在基于图的工作流中使用 Toolbox 服务的工具。LangGraph 官方指南同样适用只需做最小改动。将工具表示为节点把每个工具表示为一个 LangGraph 节点在节点功能内封装工具的执行from toolbox_langchain import ToolboxClient from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import ToolNode # Define the function that calls the model def call_model(state: MessagesState): messages state[messages] response model.invoke(messages) return {messages: [response]} # Return a list to add to existing messages model ChatVertexAI(modelgemini-3-flash-preview) builder StateGraph(MessagesState) tool_node ToolNode(tools) builder.add_node(agent, call_model) builder.add_node(tools, tool_node)这里使用MessagesState作为共享状态消息列表会在节点间自动累积ToolNode(tools)负责实际执行模型发起的工具调用。连接工具与 LLM将工具节点与 LLM 节点相连。LLM 根据输入或上下文决定使用哪个工具工具输出可以回传给 LLM 继续推理from typing import Literal from langgraph.graph import END, START from langchain_core.messages import HumanMessage # Define the function that determines whether to continue or not def should_continue(state: MessagesState) - Literal[tools, END]: messages state[messages] last_message messages[-1] if last_message.tool_calls: return tools # Route to tools node if LLM makes a tool call return END # Otherwise, stop builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue) builder.add_edge(tools, agent) graph builder.compile() graph.invoke({messages: [HumanMessage(contentDo something with the tools)]})这是一个经典的 Agent 循环agent节点调用模型 → 若模型发出tool_calls则路由到tools节点执行 → 结果回灌给agent→ 直到模型不再要求调用工具才到达END。它与仓库 Quickstart 示例quickstart.py中用create_react_agentMemorySaver的多轮对话方案互为补充——前者是 prebuilt 快速方案后者是自定义图的完全控制方案。手动调用在 Agent 框架之外你也可以用invoke方法手动执行工具适合测试工具或需要对执行过程做精确控制时使用result tools[0].invoke({name: Alice, age: 30})客户端到服务端认证Client to Server Authentication本节介绍ToolboxClient在连接一个要求认证的 Toolbox 服务实例时如何对自身进行认证。这在保障服务端点安全时至关重要尤其是部署在 Cloud Run、GKE 或任何禁止未认证访问的环境中。需要强调客户端到服务端认证与下文认证工具Authenticating Tools是不同的概念。前者在加载或调用任何工具之前就让服务端验证发起请求的客户端身份后者则是为已建立连接的 Toolbox 会话内的特定工具提供凭据。何时需要客户端到服务端认证当 Toolbox 服务配置为拒绝未认证请求时就需要此认证例如Toolbox 服务部署在 Cloud Run 上并配置为Require authentication要求认证。服务位于 Identity-Aware Proxy (IAP) 或类似的认证层之后。自托管 Toolbox 服务上有自定义认证中间件。在这些场景下如果客户端没有正确的认证配置连接或调用如load_tool很可能以Unauthorized错误失败。工作原理ToolboxClient允许你指定函数异步客户端使用协程来动态生成发往 Toolbox 服务的每个请求的 HTTP 头。最常见的用法是添加带 Bearer Token 的Authorization头例如 Google ID Token。这些头部生成函数会在每次请求前被调用确保总是使用最新的凭据或头部值。配置方式from toolbox_langchain import ToolboxClient async with ToolboxClient( toolbox-url, client_headers{header1: header1_getter, header2: header2_getter, ...} ) as client:在 Google Cloud 上认证对于托管在 Google Cloud如 Cloud Run且要求Google ID token认证的 Toolbox 服务toolbox_core.auth_methods辅助模块提供了开箱即用的工具函数aget_google_id_token异步版本与同步版本。Cloud Run 分步指南配置权限为 Cloud Run 服务的主体授予roles/run.invokerIAM 角色。主体可以是你的user account email或一个service account。配置凭据本地开发配置 Application Default Credentials (ADC)。Google Cloud 环境在 Google Cloud 内部运行时如 Compute Engine、GKE、另一个 Cloud Run 服务、Cloud FunctionsADC 通常使用环境的默认服务账号自动配置完成。连接 Toolbox 服务from toolbox_langchain import ToolboxClient from toolbox_core import auth_methods auth_token_provider auth_methods.aget_google_id_token(URL) # can also use sync method async with ToolboxClient( URL, client_headers{Authorization: auth_token_provider}, ) as client: tools client.load_toolset() # Now, you can use the client as usual.工具级认证Authenticating Tools某些工具需要用户认证才能访问敏感数据。安全提示连接应用与 Toolbox 服务时务必使用 HTTPS尤其是使用了配置过认证的工具时。使用 HTTP 会让应用面临严重的安全风险。支持的认证机制Toolbox 目前支持基于 OIDC 协议 的认证使用ID Token而非 Access Token面向 Google OAuth 2.0。配置工具首先需要在服务端把目标工具配置为要求认证——即在该工具的parameters中声明authServices把某个authService映射到 ID Token 中的特定 OIDC claim 字段。具体配置方法见 工具配置文档中的 Authenticated Parameters一个典型示例如下kind: tool name: search_flights_by_user_id type: postgres-sql source: my-pg-instance statement: | SELECT * FROM flights WHERE user_id $1 parameters: - name: user_id type: string description: Auto-populated from Google login authServices: # Refer to one of the authService defined - name: my-google-auth # sub is the OIDC claim field for user ID field: sub配置 SDK你需要在 SDK 侧提供一个从认证服务获取 ID Token 的方法async def get_auth_token(): # ... Logic to retrieve ID token (e.g., from local storage, OAuth flow) # This example just returns a placeholder. Replace with your actual token retrieval. return YOUR_ID_TOKEN # Placeholder为工具添加认证async with ToolboxClient(http://127.0.0.1:5000) as toolbox: tools toolbox.load_toolset() auth_tool tools[0].add_auth_token_getter(my_auth, get_auth_token) # Single token multi_auth_tool tools[0].add_auth_token_getters({auth_1: get_auth_1}, {auth_2: get_auth_2}) # Multiple tokens # OR auth_tools [tool.add_auth_token_getter(my_auth, get_auth_token) for tool in tools]需要注意注册 getter 时使用的名字如my_auth必须与工具配置中对应authService的name完全一致。在加载时添加认证auth_tool toolbox.load_tool(auth_token_getters{my_auth: get_auth_token}) auth_tools toolbox.load_toolset(auth_token_getters{my_auth: get_auth_token})注意加载时添加的认证 token 只影响该次调用中加载的工具。完整示例import asyncio from toolbox_langchain import ToolboxClient async def get_auth_token(): # ... Logic to retrieve ID token (e.g., from local storage, OAuth flow) # This example just returns a placeholder. Replace with your actual token retrieval. return YOUR_ID_TOKEN # Placeholder async with ToolboxClient(http://127.0.0.1:5000) as toolbox: tool toolbox.load_tool(my-tool) auth_tool tool.add_auth_token_getter(my_auth, get_auth_token) result auth_tool.invoke({input: some input}) print(result)参数绑定Parameter Binding使用 SDK 可以预先确定工具参数的值这些值不会被 LLM 修改。它的适用场景包括保护敏感信息API Key、密钥等。强制一致性确保某些参数取特定值。预填已知数据提供默认值或上下文。为已加载的工具绑定参数async with ToolboxClient(http://127.0.0.1:5000) as toolbox: tools toolbox.load_toolset() bound_tool tool[0].bind_param(param, value) # Single param multi_bound_tool tools[0].bind_params({param1: value1, param2: value2}) # Multiple params # OR bound_tools [tool.bind_param(param, value) for tool in tools]在加载时绑定参数bound_tool toolbox.load_tool(my-tool, bound_params{param: value}) bound_tools toolbox.load_toolset(bound_params{param: value})注意加载时绑定的值只影响该次调用中加载的工具。绑定动态值用一个函数绑定动态值该函数会在每次工具调用时被求值def get_dynamic_value(): # Logic to determine the value return dynamic_value dynamic_bound_tool tool.bind_param(param, get_dynamic_value)提示绑定参数值无需修改服务端的工具配置。但绑定所用参数名必须与工具配置中的参数名完全一致详见 工具配置文档 中的参数定义。安全参数Secure Parameters版本要求安全参数自toolbox-langchain1.4.0 版本起支持依赖toolbox-core 1.4.0并要求 MCP 协议版本为2026-07-28或更新同时启用com.google.cloud/toolbox.v1扩展。服务端配置细节见 Secure Parameters 配置。从仓库结构看这一扩展对应 extensions/2026-07-28/secureParams 目录含 schema 与 specification 子目录extensions/README.md 对扩展机制做了总览说明。安全参数专为敏感的运行时值设计例如终端用户的customer_id、租户标识或密钥 Token这些值不允许 LLM 看到或控制。相比普通参数安全参数具备以下能力Schema 隔离安全参数会被自动从 LangChain 的tool.args_schema中排除因此使用model.bind_tools(tools)的模型永远不会看到或请求这些参数。提示注入防御如果模型试图在标准参数中提供安全参数执行会立即失败。快速失败校验缺少必需的安全参数时会在调用前于本地直接失败。绑定方式你可以在加载工具时提供安全参数也可以把它们绑定到已加载的工具上同步与异步客户端均支持from toolbox_langchain import ToolboxClient client ToolboxClient(http://127.0.0.1:5000) # Option A: Bind secure parameters when loading tools (sync or async) bound_tool client.load_tool(search_secure_data, secure_params{customer_id: cust_12345}) tools client.load_toolset(my-set, secure_params{customer_id: cust_12345}) # Async client loading: # bound_tool await client.aload_tool(search_secure_data, secure_params{customer_id: cust_12345}) # tools await client.aload_toolset(my-set, secure_params{customer_id: cust_12345}) # Option B: Bind secure parameters to an un-bound loaded tool (returns a new immutable tool) raw_tool client.load_tool(search_secure_data) single_bound raw_tool.bind_secure_param(customer_id, cust_12345) multi_bound raw_tool.bind_secure_params({ customer_id: cust_12345, session_token: token-xyz, }) # Option C: Dynamic callable (evaluated per invocation) dynamic_tool raw_tool.bind_secure_param(customer_id, lambda: get_current_user_id())服务端配置时只需在工具参数上标记secure: true配置示例见 工具配置文档kind: tool name: search_secure_data type: postgres-sql source: my-pg-instance statement: | SELECT * FROM sessions WHERE customer_id $1 AND session_token $2 parameters: - name: customer_id type: string description: Sensitive customer identifier supplied out-of-band by the calling application secure: true - name: session_token type: string description: Sensitive session token supplied out-of-band by the calling application secure: true交叉绑定约束与互斥性为防止安全配置错误、严格区分模型参数与应用参数SDK 对绑定方法做了互斥约束在安全参数上调用tool.bind_param()会抛出ValueError: parameter name is a secure parameter; use bind_secure_param/bind_secure_params instead在普通参数上调用tool.bind_secure_param()会抛出ValueError: parameter name is a regular parameter; use bind_param/bind_params instead异步使用为了通过协作式多任务获得更好的性能可以使用ToolboxClient的异步接口import asyncio from toolbox_langchain import ToolboxClient async def main(): async with ToolboxClient(http://127.0.0.1:5000) as toolbox: tool await client.aload_tool(my-tool) tools await client.aload_toolset() response await tool.ainvoke() if __name__ __main__: asyncio.run(main())注意aload_tool、aload_toolset等异步接口需要异步运行环境。运行异步 Python 程序的指导见 Python 官方 asyncio 文档。此前 Quickstart 与仓库示例quickstart.py中的async with写法正是异步接口的典型用法。OpenTelemetry 可观测性SDK 通过toolbox-core层支持 OpenTelemetry 追踪与指标遵循 MCP Semantic Conventions。启用后每次tools/list与tools/call都会产生客户端 span并记录操作耗时直方图。安装与启用首先安装toolbox-core的 telemetry 附加依赖pip install toolbox-core[telemetry]然后在创建客户端时传入telemetry_enabledTruefrom toolbox_langchain import ToolboxClient with ToolboxClient(http://127.0.0.1:5000, telemetry_enabledTrue) as toolbox: tool toolbox.load_tool(my-tool) result tool.invoke({param: value})在创建客户端之前需要先配置好 OpenTelemetry 的TracerProvider与MeterProvider完整配置示例见 Core SDK OpenTelemetry 章节。每次调用的遥测属性使用TelemetryAttributes为工具调用附加模型、用户与 Agent 元数据from toolbox_core import TelemetryAttributes from toolbox_langchain import ToolboxClient attrs TelemetryAttributes( llm_modelgemini-3.6-flash, user_iduser-123, agent_idagent-abc, ) with ToolboxClient(http://127.0.0.1:5000) as toolbox: tools toolbox.load_toolset(my-toolset, telemetry_attributesattrs) tool toolbox.load_tool(my-tool) instrumented_tool tool.add_telemetry_attributes(attrs)你可以把telemetry_attributes传给load_tool()或load_toolset()也可以在已加载的工具上调用add_telemetry_attributes()。从 Core SDK 文档 可知其底层行为这些属性会通过 MCP 请求的params._meta中dev.mcp-toolbox/telemetry键发送给 Toolbox 服务端可供 SQL Commenter 等数据库工具的服务端插桩使用同时在启用遥测时作为客户端 span 的属性被记录TelemetryAttributes的三个可选字段llm_model、user_id、agent_id会分别序列化为client.model、client.user.id、client.agent.id键。此外add_telemetry_attributes()同样遵循不可变模式返回新工具实例重复调用会替换而非合并之前的属性未设置字段与空字符串会在发送前被丢弃。总结toolbox-langchain把 MCP Toolbox 的数据库工具能力无缝桥接到 LangChain / LangGraph 生态通过ToolboxClient连接服务、load_toolset()/load_tool()加载工具、create_react_agent或自定义StateGraph编排 Agent、invoke()手动执行生产环境中则依赖protocol版本协商、client_headers客户端认证、auth_token_getters工具认证、bind_param(s)参数绑定、secure_params安全参数与 OpenTelemetry 遥测来满足安全与可观测性要求。结合本仓库中的 Quickstart 教程、完整示例、工具配置文档 以及服务端 MCP 协议实现 与 扩展定义你可以从零开始构建并逐步加固一个生产可用的数据库 Agent。【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考