企业官网建设流程全解析

用 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 可观测性等生产级配置，并能在当前仓库中找到对应的服务端实现与完整示例代码作为佐证。

Overview：toolbox-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 = """ You're a helpful hotel assistant. You handle hotel searching, booking and cancellations. When the user searches for a hotel, mention it's 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. Don't 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(model="gemini-2.5-flash") # model = ChatAnthropic(model="claude-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, checkpointer=MemorySaver()) 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_mode="values", config=config) print(f"[OUTPUT] AI: {response['messages'][-1].content}") asyncio.run(main())

这个示例展示了三个关键点：ToolboxClient作为异步上下文管理器使用、aload_toolset()一次性加载整个工具集、以及MemorySaver让 Agent 在多轮对话中保持状态（这正是 LangGraph 相比普通 LangChain 调用的核心优势）。

Quickstart：最小可运行示例

官方文档给出了一个最精简的入门示例，使用 LangGraph 的create_react_agent构建 Agent：

from 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(model="gemini-3-flash-preview") agent = create_react_agent(model, tools) prompt = "How's the weather today?" for s in agent.stream({"messages": [("user", prompt)]}, stream_mode="values"): 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_mode="values")逐轮输出消息，pretty_print()负责格式化打印。

完整的多轮对话版本同样见 quickstart.py。如果需要在构建 Agent 前先完整掌握服务端搭建，请先阅读 Toolbox Quickstart 教程。

初始化客户端与传输协议

基础初始化

导入并初始化客户端，指向正在运行的 Toolbox 服务：

from toolbox_langchain import ToolboxClient # Replace with your Toolbox service's 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实现的。

从源码结构看，这些协议版本并不是虚构的：仓库的服务端 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", protocol=Protocol.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", protocol=Protocol.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(model="gemini-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(model="gemini-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(content="Do something with the tools")]})

这是一个经典的 Agent 循环：agent节点调用模型 → 若模型发出tool_calls则路由到tools节点执行 → 结果回灌给agent→ 直到模型不再要求调用工具才到达END。它与仓库 Quickstart 示例（quickstart.py）中用create_react_agent+MemorySaver的多轮对话方案互为补充——前者是 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 分步指南

1. 配置权限：为 Cloud Run 服务的主体授予roles/run.invokerIAM 角色。主体可以是你的user account email或一个service account。
2. 配置凭据：
   • 本地开发：配置 Application Default Credentials (ADC)。
   • Google Cloud 环境：在 Google Cloud 内部运行时（如 Compute Engine、GKE、另一个 Cloud Run 服务、Cloud Functions），ADC 通常使用环境的默认服务账号自动配置完成。
3. 连接 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_enabled=True：

from toolbox_langchain import ToolboxClient with ToolboxClient("http://127.0.0.1:5000", telemetry_enabled=True) 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_model="gemini-3.6-flash", user_id="user-123", agent_id="agent-abc", ) with ToolboxClient("http://127.0.0.1:5000") as toolbox: tools = toolbox.load_toolset("my-toolset", telemetry_attributes=attrs) 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