企业官网建设流程全解析

MCP Toolbox Python Core SDK（toolbox-core）实战指南：工具加载、协议协商、安全参数与可观测性

【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox

本指南以 MCP Toolbox for Databases 的官方 Python 核心 SDK（toolbox-core）为主题，系统讲解如何在自有 Agent 应用中加载、调用 MCP Toolbox 服务托管的数据库工具，涵盖传输协议协商、同步/异步客户端、LangGraph 集成、两层认证体系（客户端到服务器、工具级认证）、参数绑定与安全参数（Secure Parameters）以及 OpenTelemetry 可观测性。读完本文，你将能够独立搭建「Toolbox 服务 + Python 应用」的完整链路，并掌握生产环境下敏感参数的防护与调用链追踪方案。

概览：SDK 与 Toolbox 服务的关系

toolbox-core是 MCP Toolbox 的官方 Python 核心包，它为运行中的 MCP Toolbox 服务提供纯 Python 接口，使你的应用能够**加载（load）并调用（invoke）**服务中定义的各类数据库工具。

理解二者关系至关重要：MCP Toolbox 本身是一个开源的 Model Context Protocol（MCP）服务器，位于你的应用编排框架与数据库之间，集中管理工具的定义、分发与调用（架构与整体说明见 Introduction）。而toolbox-core则是应用侧的客户端 SDK，它通过网络与 Toolbox 服务通信——因此使用本 SDK 的前提是有一个正在运行的 Toolbox 服务。典型链路为：

你的 Agent 应用（Python）──(MCP over HTTP)──> MCP Toolbox 服务 ──> 数据库

从仓库源码可以印证这种「多协议版本并存」的服务端设计：在 internal/server/mcp/mcp.go 中，ProcessMethod按协商的 MCP 版本分发到v20241105、v20250326、v20250618、v20251125、v20260728五个协议处理器，这与下文 SDK 侧Protocol枚举支持的版本一一对应。

安装

安装toolbox-core只需一条命令：

pip install toolbox-core

注意：默认的ToolboxClient是异步客户端，加载与调用工具均需使用await（如大多数示例所示）。异步代码必须运行在事件循环内（例如通过asyncio.run()或某个异步框架），相关细节可参考 Python 官方asyncio文档。如果你偏好同步执行，见下文「同步用法」小节。

注意：ToolboxClient（以及其同步版本ToolboxSyncClient）通过底层的 HTTP 客户端会话与网络资源交互。请务必使用上下文管理器或显式调用close()来释放这些资源。如果你自行提供了 session，则除了调用ToolboxClient.close()之外，还需要自行关闭你提供的 session。

若需要使用 OpenTelemetry 遥测能力，可安装带 extra 的版本（详见「OpenTelemetry」小节）：

pip install toolbox-core[telemetry]

快速开始

第一步：启动 Toolbox 服务

确保本地机器的5000端口上运行着 MCP Toolbox 服务。你可以通过以下任一方式启动：

• 使用 npx（适合快速体验）：npx @toolbox-sdk/server --prebuilt=postgres，然后在 MCP 客户端配置中指向它；
• 使用二进制或容器镜像：下载对应平台的toolbox可执行文件后运行./toolbox --config "tools.yaml"（工具的 YAML 配置方法见 Tools 配置文档）；
• 使用 Homebrew：brew install mcp-toolbox后直接执行toolbox --config "tools.yaml"。

详细的端到端教程（含数据库准备、服务配置与各框架 Agent 接入）见 Toolbox Quickstart Tutorial。

第二步：最小示例

import asyncio from toolbox_core import ToolboxClient async def main(): # Replace with the actual URL where your Toolbox service is running async with ToolboxClient("http://127.0.0.1:5000") as toolbox: weather_tool = await toolbox.load_tool("get_weather") result = await weather_tool(location="London") print(result) if __name__ == "__main__": asyncio.run(main())

资源清理

如果你在初始化ToolboxClient时没有提供外部 session，又无法使用async with，则必须显式关闭客户端，以确保内部创建的 session 被释放：

toolbox = ToolboxClient("http://127.0.0.1:5000") try: # ... use toolbox ... finally: await toolbox.close()

客户端初始化与生命周期

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

from toolbox_core import ToolboxClient # Replace with your Toolbox service's URL async with ToolboxClient("http://127.0.0.1:5000") as toolbox:

所有与加载、调用工具相关的交互都通过这个客户端完成。几个关键生命周期规则：

• 外部 session：高级场景下，你可以在初始化时传入外部的aiohttp.ClientSession（如ToolboxClient(url, session=my_session)）。一旦提供自己的 session，你就需要负责其生命周期，ToolboxClient不会关闭它。
• 关闭客户端的影响：关闭ToolboxClient会同时关闭该客户端下所有工具共享的底层网络会话。因此，任何已加载的工具实例在客户端关闭后将失效，若此时尝试调用会抛出错误。
• URL 查询参数保留：如果你的连接 URL 包含查询参数（例如http://localhost:5000?foo=bar），客户端会在所有 API 请求中自动保留它们，用于参数绑定（对应服务端文档中的 URL Parameter Binding）。

传输协议与协议协商

SDK 支持多种传输协议与 Toolbox 服务器通信。默认情况下，客户端使用最新支持的 Model Context Protocol（MCP）版本。

你可以在初始化客户端时通过protocol选项显式选择协议。这在需要使用 Toolbox 原生 HTTP 协议、或将客户端固定到某个特定旧版 MCP 时非常有用。

说明：MCP 传输均基于Model Context Protocol over HTTP。

支持的协议版本

SDK 目前支持多个版本的 MCP 协议，Protocol枚举的完整定义位于toolbox_core/protocol模块（Protocol枚举）。

从仓库源码看，服务端对这些版本均有对应实现：在 internal/server/mcp/mcp.go 中，2024-11-05（以及未协商协议时的默认值）由v20241105处理器处理，其余版本按字面量分发到各自处理器；Introduction 文档 同样列出了这五个稳定协议版本，并说明服务器在未协商协议时默认回退到2024-11-05的旧式传输。

示例：显式指定协议

from toolbox_core 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

如果你希望将首选起始协议设为2025-03-26（允许在服务器不支持时回退协商）：

from toolbox_core 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

若要限制协商到特定的版本子集，可向protocol参数传入协议列表：

from toolbox_core import ToolboxClient from toolbox_core.protocol import Protocol async with ToolboxClient( "http://127.0.0.1:5000", protocol=[Protocol.MCP_LATEST, Protocol.MCP_v20250618] ) as toolbox: # Use client pass

提示：如果你希望严格固定版本、禁用协议回退，必须传入只包含单个值的数组：protocol=[Protocol.MCP_DRAFT]。

加载工具

你可以单独加载工具，也可以按工具组（toolset，即你在 Toolbox 服务配置中定义的工具集合）批量加载。加载 toolset 适合处理多个相关函数，单独加载单个工具则提供更细粒度的控制。

加载工具组（toolset）

toolset 是一组相关工具的集合，可以加载其中全部工具或指定某个 toolset：

# Load all tools tools = await toolbox.load_toolset() # Load a specific toolset tools = await toolbox.load_toolset("my-toolset")

加载单个工具

按工具的唯一名称加载指定工具，实现细粒度控制：

tool = await toolbox.load_tool("my-tool")

补充说明：工具与 toolset 本身是在 Toolbox 服务的配置文件（如tools.yaml）中以kind: tool声明的，包含名称、类型（如postgres-sql）、数据源、SQL 语句、参数定义等字段，完整配置规范见 Tools 配置文档。SDK 的load_tool/load_toolset本质上是通过 MCPtools/list协议从服务端拉取这些工具清单并构造可调用对象。

调用工具

加载完成后，工具表现为可await的 Python 函数。使用await调用它们，并传入与 Toolbox 服务中该工具配置所定义参数相对应的实参：

tool = await toolbox.load_tool("my-tool") result = await tool("foo", bar="baz")

提示：关于如何完整搭建你需要运行的 Toolbox 服务本身，请参考 Toolbox Quickstart Guide。

同步用法

默认情况下，ToolboxClient及其产生的ToolboxTool对象行为类似异步 Python 函数，需要使用await。

如果你的应用以同步代码为主，或不想管理 asyncio 事件循环，可以使用 SDK 提供的同步替代方案：

• ToolboxSyncClient：ToolboxClient的同步对应物。
• ToolboxSyncTool：ToolboxTool的同步对应物。

ToolboxSyncClient同步处理与 Toolbox 服务的通信，并在加载工具时产生ToolboxSyncTool实例。使用这些同步版本时不需要await关键字：

from toolbox_core import ToolboxSyncClient with ToolboxSyncClient("http://127.0.0.1:5000") as toolbox: weather_tool = toolbox.load_tool("get_weather") result = weather_tool(location="Paris") print(result)

提示：虽然为方便起见提供了同步调用，但对于工具调用这类 I/O 密集任务，通常的最佳实践是使用异步操作（即默认的ToolboxClient与ToolboxTool）。异步编程允许协作式多任务，在处理并发请求的应用中往往能带来更好的性能与资源利用率。

与 LangGraph 集成

Toolbox Core SDK 可与 LangGraph 等框架平滑集成，让你把 Toolbox 服务托管的工具纳入 Agent 工作流。

提示：加载出的工具（异步的ToolboxTool与同步的ToolboxSyncTool）本身即可调用，通常能直接使用。但为确保 Google 风格 docstring 中的参数描述能被准确解析并供 LLM（通过bind_tools()）及 LangGraph 内部使用，建议使用 LangChain 的StructuredTool包装这些工具。

以下是一个概念性示例（改编自官方 LangGraph tool calling 指南）：

import asyncio from typing import Annotated from typing_extensions import TypedDict from langchain_core.messages import HumanMessage, BaseMessage from toolbox_core import ToolboxClient from langchain_google_vertexai import ChatVertexAI from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain.tools import StructuredTool from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list[BaseMessage], add_messages] async def main(): async with ToolboxClient("http://127.0.0.1:5000") as toolbox: tools = await toolbox.load_toolset() wrapped_tools = [StructuredTool.from_function(tool, parse_docstring=True) for tool in tools] model_with_tools = ChatVertexAI(model="gemini-3-flash-preview").bind_tools(wrapped_tools) tool_node = ToolNode(wrapped_tools) def call_agent(state: State): response = model_with_tools.invoke(state["messages"]) return {"messages": [response]} def should_continue(state: State): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END graph_builder = StateGraph(State) graph_builder.add_node("agent", call_agent) graph_builder.add_node("tools", tool_node) graph_builder.add_edge(START, "agent") graph_builder.add_conditional_edges( "agent", should_continue, ) graph_builder.add_edge("tools", "agent") app = graph_builder.compile() prompt = "What is the weather in London?" inputs = {"messages": [HumanMessage(content=prompt)]} print(f"User: {prompt}\n") print("--- Streaming Agent Steps ---") events = app.stream( inputs, stream_mode="values", ) for event in events: event["messages"][-1].pretty_print() print("\n---\n") asyncio.run(main())

客户端到服务器认证（Client-to-Server Authentication）

本小节描述如何让ToolboxClient自身在连接需要认证的 MCP Toolbox 服务器时完成身份认证。这对于保护你的 Toolbox 服务端点至关重要，尤其是部署在 Cloud Run、GKE 或任何限制未认证访问的环境中。

这种客户端到服务器的认证确保 Toolbox 服务器在加载或调用任何工具之前，就能校验请求方的身份。它不同于下文「Authenticating Tools（工具认证）」——后者处理的是在已连接的 Toolbox 会话内为特定工具提供凭据。

何时需要客户端到服务器认证

当你的 Toolbox 服务器被配置为拒绝未认证请求时需要此认证，例如：

• Toolbox 服务器部署在 Cloud Run 且配置为「Require authentication（要求认证）」；
• 服务器位于 Identity-Aware Proxy（IAP）或类似的认证层之后；
• 自托管 Toolbox 服务器上挂载了自定义认证中间件。

在这些场景下，如果没有正确的客户端认证，连接或发起调用（如load_tool）通常会以Unauthorized错误失败。

工作原理

ToolboxClient（以及ToolboxSyncClient）允许你指定函数（异步客户端用协程）为发送给 Toolbox 服务器的每个请求动态生成 HTTP 头。最常见的用例是添加携带 bearer token（如 Google ID token）的Authorization头。

这些生成头的函数在每个请求发出前被调用，确保可以使用最新的凭据或头值。

配置方式

按如下方式配置这些动态请求头：

from toolbox_core import ToolboxClient async with ToolboxClient("toolbox-url", client_headers={"header1": header1_getter, "header2": header2_getter, ...}) as client: # Use client pass

认证 Google Cloud 服务器

对于托管在 Google Cloud（如 Cloud Run）且要求Google ID token认证的 Toolbox 服务器，SDK 提供了辅助模块toolbox_core.auth_methods（auth_methods模块），内含获取 ID token 的工具函数。

Cloud Run 逐步操作指南

1. 配置权限：为 Cloud Run 服务的主体授予roles/run.invokerIAM 角色。该主体可以是你的用户账号邮箱或一个服务账号。
2. 配置凭据：
   • 本地开发：按 ADC（Application Default Credentials）规范完成本地开发环境设置。
   • Google Cloud 环境：当在 Google Cloud 内运行（如 Compute Engine、GKE、另一个 Cloud Run 服务、Cloud Functions）时，ADC 通常自动配置，使用环境的默认服务账号。
3. 连接 Toolbox 服务器：

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 = await client.load_toolset() # Now, you can use the client as usual.

工具认证（Authenticating Tools）

重要：在生产环境或任何涉及敏感数据（包括工具需要认证令牌的场景）的通信中，请始终使用 HTTPS连接应用与 Toolbox 服务。使用纯 HTTP 缺少加密，会使你的应用和数据暴露于窃听、篡改等重大安全风险之中。

工具可以在 Toolbox 服务内被配置为需要认证，以确保只有授权的用户或应用才能调用它们，尤其是在访问敏感数据时。

何时需要认证

认证是在 Toolbox 服务内按工具（per-tool）配置的。如果你要使用的工具在服务中被标记为需要认证，则必须配置 SDK 客户端，在调用该特定工具时提供所需凭据（目前为 OAuth2 token）。

支持的认证机制

Toolbox 服务通过 **Authenticated Parameters（认证参数）**实现安全的工具使用。关于这些机制在 Toolbox 服务中的工作原理与配置方法，请参见 Authenticated Parameters 配置文档。简言之，你需要在工具配置中把某个参数映射到authService定义的 OIDC claim（如sub），该参数便会在请求时由服务端从 ID token 自动填充，而不需要模型提供。

步骤 1：在 Toolbox 服务中配置工具

首先，确保目标工具在 Toolbox 服务中已正确配置为需要认证，配置指引见上述 Authenticated Parameters。

步骤 2：配置 SDK 客户端

你的应用需要一种方式为已认证用户获取所需的 OAuth2 token。SDK 要求你提供一个在工具被调用时能够获取该 token 的函数。

提供 ID Token 获取函数

你必须向 SDK 提供一个（同步或异步）函数，调用时返回所需 token。具体实现取决于你应用的认证流程（例如从存储中读取 token、发起 OAuth 流程等）。

重要：向 SDK 注册 getter 函数时使用的名称（例如"my_api_token"）必须与 Toolbox 服务中该工具配置里对应authService的name完全一致。

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

提示：你的 token 获取函数会在每次认证参数需要 token 的工具调用时被调用。请考虑在该函数内实现缓存逻辑，避免重复获取或生成 token，尤其对于有效期较长或获取过程资源密集的 token。

选项 A：为已加载工具添加认证

你可以在工具加载之后为其添加 token 获取函数。这会修改该特定工具实例。

async with ToolboxClient("http://127.0.0.1:5000") as toolbox: tool = await toolbox.load_tool("my-tool") auth_tool = tool.add_auth_token_getter("my_auth", get_auth_token) # Single token # OR multi_auth_tool = tool.add_auth_token_getters({ "my_auth_1": get_auth_token_1, "my_auth_2": get_auth_token_2, }) # Multiple tokens

选项 B：加载工具时添加认证

你也可以在load_tool或load_toolset调用中直接提供 token 获取函数。这种方式仅对本次调用加载的工具生效，不会修改之前已加载的工具对象。

auth_tool = await toolbox.load_tool(auth_token_getters={"my_auth": get_auth_token}) # OR auth_tools = await toolbox.load_toolset(auth_token_getters={"my_auth": get_auth_token})

注意：加载时添加的认证 token 只影响本次调用中加载的工具。

完整认证示例

import asyncio from toolbox_core 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 = await toolbox.load_tool("my-tool") auth_tool = tool.add_auth_token_getters({"my_auth": get_auth_token}) result = auth_tool(input="some input") print(result)

注意：某个名称（例如"GOOGLE_ID"）的认证 token getter 会替换掉同名加"_token"后缀的客户端请求头（例如"GOOGLE_ID_token"）。

参数绑定（Parameter Binding）

SDK 允许你在工具被调用、甚至传给 LLM 之前，为特定工具参数**预置（绑定）**值。这些绑定值是固定的，在工具使用过程中不会被 LLM 请求或修改。

为什么要绑定参数

• 保护敏感信息：API 密钥、机密等。
• 强制一致性：确保某些参数取特定值。
• 预填已知数据：提供默认值或上下文。

重要：用于绑定的参数名（如"api_key"）必须与 Toolbox 服务中该工具配置定义的参数名完全一致。

注意：使用 SDK 绑定参数值无需修改Toolbox 服务中的工具配置。

选项 A：为已加载工具绑定参数

在工具加载之后为其绑定值。这会修改该特定工具实例。

async with ToolboxClient("http://127.0.0.1:5000") as toolbox: tool = await toolbox.load_tool("my-tool") bound_tool = tool.bind_param("param", "value") # OR bound_tool = tool.bind_params({"param": "value"})

选项 B：加载工具时绑定参数

在加载工具时直接指定绑定参数。这种方式仅对本次调用加载的工具生效。

bound_tool = await toolbox.load_tool("my-tool", bound_params={"param": "value"}) # OR bound_tools = await toolbox.load_toolset(bound_params={"param": "value"})

绑定动态值

除了静态值，你还可以把参数绑定到同步或异步函数。该函数会在每次调用工具时被执行，从而在运行时动态确定参数值。

注意：绑定参数值同样无需修改工具配置。

async def get_dynamic_value(): # Logic to determine the value return "dynamic_value" # Assuming `tool` is a loaded tool instance from a ToolboxClient dynamic_bound_tool = tool.bind_param("param", get_dynamic_value)

安全参数（Secure Parameters）

前置条件：安全参数自toolbox-core1.4.0版本起支持，要求 MCP 协议版本2026-07-28 或更新，并且服务器启用了com.google.cloud/toolbox.v1扩展。服务器端配置细节见 Secure Parameters 配置文档 与 扩展说明。

安全参数专为敏感的运行时上下文设计——例如终端用户的customer_id、租户标识或秘密令牌——这些值不能让 LLM 看到、控制或臆造。

与普通参数不同，在 Toolbox 服务器配置中标记为secure: true的参数具有以下关键特性：

• Schema 隔离（Schema Isolation）：SDK 会完全剥离安全参数在公开工具声明、docstring 和运行时函数签名（反映在__signature__与inspect.signature(tool)）中的存在。LLM 永远不会感知这些参数，从而保持模型上下文窗口干净并防止凭据泄露。
• 提示注入防御（Prompt Injection Defense）：如果模型或调用方试图在标准参数中为安全参数提供值，SDK 会立即拒绝执行。
• 快速失败校验（Fast-Fail Validation）：SDK 会在请求发出前本地校验所有必需的安全参数是否已绑定。若缺少任何必需安全参数，执行会立即失败。
• 线协议分离（Wire Protocol Separation）：安全参数通过 MCP 2026-07-28tools/callJSON-RPC 载荷中的secureArguments字段带外传输，与普通参数完全隔离。

服务器端实现佐证

仓库源码印证了上述设计：在 internal/server/mcp/v20260728/manifests.go 的generateToolManifest中，参数按GetSecure()被拆分为普通参数与安全参数两组，普通参数进入inputSchema，安全参数单独生成secureInputSchema放入工具清单；同时 GenerateListToolsResult 会跳过「定义了安全参数但客户端未声明扩展支持」的工具，防止不支持的调用。完整的协议规范（包括tools/call中secureArguments的载荷结构、错误码-32021/-32602与各类错误矩阵）见 Secure Parameters 规范。扩展标识符com.google.cloud/toolbox.v1的版本化策略见 Extensions 总览，服务器可用--disable-ext com.google.cloud/toolbox.v1标志禁用该扩展。

对应地，服务端工具配置中把参数标记为安全参数的 YAML 形如（完整配置见 Tools 配置文档）：

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

选项 A：为已加载工具绑定安全参数

在工具加载之后绑定安全值。每个绑定方法都会返回一个新的、不可变的工具实例，原工具保持不变。

from toolbox_core import ToolboxClient async with ToolboxClient("http://127.0.0.1:5000") as toolbox: tool = await toolbox.load_tool("search_secure_data") # Bind a single secure parameter bound_tool = tool.bind_secure_param("customer_id", "cust_12345") # OR bind multiple secure parameters at once multi_bound_tool = tool.bind_secure_params({ "customer_id": "cust_12345", "session_token": "token-xyz" })

选项 B：加载工具时绑定安全参数

在加载工具或工具组时预绑定安全参数。SDK 会校验所提供键在目标工具上均为已存在的安全参数。

async with ToolboxClient("http://127.0.0.1:5000") as toolbox: # Load a single tool with secure parameters tool = await toolbox.load_tool( "search_secure_data", secure_params={"customer_id": "cust_12345"} ) # Load an entire toolset with secure parameters tools = await toolbox.load_toolset( "my-toolset", secure_params={"customer_id": "cust_12345"} )

绑定动态安全值

你也可以将安全参数绑定到同步或异步可调用对象。该可调用对象会在每次调用工具时于执行期被求值：

async def get_current_user_token() -> str: # Dynamically fetch session token or user identifier return "session-token-abc" secure_tool = tool.bind_secure_param("auth_token", get_current_user_token)

同步用法

安全参数绑定同样适用于ToolboxSyncClient下的ToolboxSyncTool：

from toolbox_core import ToolboxSyncClient with ToolboxSyncClient("http://127.0.0.1:5000") as toolbox: tool = toolbox.load_tool( "search_secure_data", secure_params={"customer_id": "cust_12345"} ) result = tool()

交叉绑定的指导与互斥性

为防止安全配置错误，并确保模型参数与应用参数的严格分离：

• 对安全参数调用bind_param()或bind_params()会抛出：ValueError: parameter '<name>' is a secure parameter; use bind_secure_param/bind_secure_params instead
• 对普通参数调用bind_secure_param()或bind_secure_params()会抛出：ValueError: parameter '<name>' is a regular parameter; use bind_param/bind_params instead

这也与服务端配置约束一致：安全参数始终是必需的，不能设为可选；一个参数不能同时设置secure: true与authServices、default或required: false（见 Tools 配置文档）。

OpenTelemetry 可观测性

SDK 支持遵循MCP Semantic Conventions的 OpenTelemetry 追踪与指标。启用后，每次tools/list与tools/call操作都会产生客户端 span 并记录操作耗时直方图，同时 W3Ctraceparent/tracestate头会被传播到 Toolbox 服务器，用于分布式追踪。

安装

安装telemetryextra 以引入 OpenTelemetry 依赖：

pip install toolbox-core[telemetry]

启用

创建ToolboxClient或ToolboxSyncClient时传入telemetry_enabled=True：

from toolbox_core import ToolboxClient async with ToolboxClient("http://127.0.0.1:5000", telemetry_enabled=True) as toolbox: tool = await toolbox.load_tool("my-tool") result = await tool(param="value")

配置 OpenTelemetry Provider

SDK 读取全局配置的TracerProvider与MeterProvider。请在创建客户端之前于应用中完成配置：

from opentelemetry import trace, metrics from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader # Configure tracing tracer_provider = TracerProvider() tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(tracer_provider) # Configure metrics metric_reader = PeriodicExportingMetricReader(OTLPMetricExporter()) meter_provider = MeterProvider(metric_readers=[metric_reader]) metrics.set_meter_provider(meter_provider) # Now create the client with telemetry enabled async with ToolboxClient("http://127.0.0.1:5000", telemetry_enabled=True) as toolbox: ...

注意：如果telemetry_enabled=True但未配置任何 provider，将使用 OpenTelemetry 的 no-op 实现——不会导出任何数据且零开销。必须安装可选的[telemetry]extra，telemetry_enabled=True才会生效；若未安装该 extra，该标志会被静默忽略。

单次调用级遥测属性

除了telemetry_enabled=True启用的自动插桩，你还可以为单个工具附加遥测属性（如 LLM 模型名、用户 ID 或 Agent ID）到外发调用上。这些属性会：

• 在 MCP 请求params._meta的dev.mcp-toolbox/telemetry键下发送给 Toolbox 服务器，供服务端插桩使用（例如数据库工具上的 SQL Commenter）；
• 在启用遥测时作为调用对应客户端 OpenTelemetry span 上的属性被记录。

使用TelemetryAttributes模型与已加载工具上的add_telemetry_attributes()方法：

from toolbox_core import ToolboxClient, TelemetryAttributes async with ToolboxClient("http://127.0.0.1:5000", telemetry_enabled=True) as toolbox: tool = await toolbox.load_tool("my-tool") attrs = TelemetryAttributes( llm_model="gemini-2.5-pro", user_id="user-123", agent_id="agent-abc", ) instrumented_tool = tool.add_telemetry_attributes(attrs) result = await instrumented_tool(param="value")

同样的方法也适用于同步场景下的ToolboxSyncTool。

字段与线上映射

TelemetryAttributes暴露三个可选字段，它们在线上序列化为 OpenTelemetry 风格的键：

提示：

• add_telemetry_attributes()返回一个新的工具实例，原工具保持不变（与bind_param、add_auth_token_getter相同的不可变模式）。
• 第二次调用add_telemetry_attributes()会替换之前的属性，而不是合并。请传入一个包含全部所需字段的TelemetryAttributes对象。
• 未设置的字段与空字符串在发送前会被丢弃，因此不会以空值出现在服务器端。

小结与最佳实践

• 先起服务，再连 SDK：toolbox-core的一切能力都建立在运行中的 Toolbox 服务之上，工具清单来自服务端配置（Tools 配置文档）。
• 管理好资源生命周期：优先使用async with或with；自行提供 session 时记得额外关闭；关闭客户端会使所有已加载工具失效。
• 按需协商协议：默认使用最新 MCP 版本；需要兼容旧客户端时可传入协议列表参与回退协商，需要严格固定时可传入单元素数组禁用回退。
• 敏感值分层防护：客户端到服务器认证保护「谁在连接服务」，工具认证保护「谁在调用工具」，安全参数保护「LLM 永远不该看到的值」；生产环境务必使用 HTTPS。
• 为生产构建可观测性：启用telemetry_enabled=True并配置 provider，再结合TelemetryAttributes在调用级关联 LLM 模型、用户与 Agent，即可实现端到端追踪。

【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox