用 MCP Toolbox 快速搭建 BigQuery 酒店预订智能体本地端到端上手指南【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolboxMCP Toolbox for Databases下称 Toolbox是一个面向数据库的开源 MCP Server可以将 BigQuery 等数据源封装成可供 LLM 直接调用的工具。本文以「酒店搜索 / 预订 / 改期 / 取消」为例完整演示在本地环境从零构建一个可运行的大模型智能体Agent先通过bqCLI 准备 BigQuery 数据集与表再编写tools.yaml将表操作声明为bigquery-sql工具并启动 Toolbox Server最后分别用 Toolbox Core SDK、LangChain、LlamaIndex 与 Google ADK 四种方式把 Agent 接到 Toolbox 上执行真实查询。读完本文你将掌握 Toolbox 与 BigQuery 集成的最小可运行闭环以及bigquery数据源与bigquery-sql工具的配置要点。本文内容以仓库中的 local_quickstart.md 为骨架并结合 BigQuery 数据源文档、bigquery-sql 工具文档、bigquery-sql 实现源码 与 BigQuery 预置配置 进行源码级补充。一、开始之前环境准备本文假设你已经完成以下准备工作安装 Python 3.10并确保pip可用建议使用venv等虚拟环境工具管理依赖。安装并配置 Google Cloud SDKgcloudCLI。通过 ADCApplication Default Credentials完成身份认证gcloud auth login --update-adc设置默认 GCP 项目将YOUR_PROJECT_ID替换为你的真实项目 IDgcloud config set project YOUR_PROJECT_ID export GOOGLE_CLOUD_PROJECTYOUR_PROJECT_IDToolbox 与各客户端库会默认使用该项目操作 BigQuery除非在配置中被显式覆盖。在 GCP 项目中启用 BigQuery API。安装 BigQuery 的 Python 客户端库pip install google-cloud-bigquery按所选框架完成 LLM 接入准备Core / LangChain安装langchain-vertexai、langchain-google-genai或langchain-anthropic中的对应包LlamaIndex安装llama-index-llms-google-genai或llama-index-llms-anthropicADK安装google-adk包。从源码看Toolbox 的 BigQuery 集成默认依赖 Application Default Credentials 完成认证。相关字段如scopes、useClientOAuth、impersonateServiceAccount在 BigQuery 数据源实现 的Config结构体中有明确定义稍后会在配置章节详解。二、Step 1在 BigQuery 中创建数据集与表先为 Agent 准备一份需要访问的测试数据。所有 BigQuery 操作都基于你在前面配置好的 Google Cloud 项目执行。2.1 创建数据集export BQ_DATASET_NAMEYOUR_DATASET_NAME # 例如 toolbox_ds export BQ_LOCATIONUS # 例如 US、EU、asia-northeast1 bq --location$BQ_LOCATION mk $BQ_DATASET_NAME也可以通过 Google Cloud Console 的 BigQuery 页面完成同样的操作。权限提示在生产环境中请确保运行 Toolbox 的服务账号或用户对数据集/项目具备必要的 IAM 权限例如 BigQuery Data Editor、BigQuery User。本地快速开始使用用户凭据你的自有权限即生效。2.2 创建 hotels 表新建文件create_hotels_table.sql内容如下CREATE TABLE IF NOT EXISTS YOUR_PROJECT_ID.YOUR_DATASET_NAME.hotels ( id INT64 NOT NULL, name STRING NOT NULL, location STRING NOT NULL, price_tier STRING NOT NULL, checkin_date DATE NOT NULL, checkout_date DATE NOT NULL, booked BOOLEAN NOT NULL );注意请把 SQL 中的YOUR_PROJECT_ID与YOUR_DATASET_NAME替换成真实值。然后执行该 SQLbq query --project_id$GOOGLE_CLOUD_PROJECT --dataset_id$BQ_DATASET_NAME --use_legacy_sqlfalse create_hotels_table.sql2.3 插入初始数据新建文件insert_hotels_data.sql加入如下 INSERT 语句同样替换项目与数据集占位符INSERT INTO YOUR_PROJECT_ID.YOUR_DATASET_NAME.hotels (id, name, location, price_tier, checkin_date, checkout_date, booked) VALUES (1, Hilton Basel, Basel, Luxury, 2024-04-20, 2024-04-22, FALSE), (2, Marriott Zurich, Zurich, Upscale, 2024-04-14, 2024-04-21, FALSE), (3, Hyatt Regency Basel, Basel, Upper Upscale, 2024-04-02, 2024-04-20, FALSE), (4, Radisson Blu Lucerne, Lucerne, Midscale, 2024-04-05, 2024-04-24, FALSE), (5, Best Western Bern, Bern, Upper Midscale, 2024-04-01, 2024-04-23, FALSE), (6, InterContinental Geneva, Geneva, Luxury, 2024-04-23, 2024-04-28, FALSE), (7, Sheraton Zurich, Zurich, Upper Upscale, 2024-04-02, 2024-04-27, FALSE), (8, Holiday Inn Basel, Basel, Upper Midscale, 2024-04-09, 2024-04-24, FALSE), (9, Courtyard Zurich, Zurich, Upscale, 2024-04-03, 2024-04-13, FALSE), (10, Comfort Inn Bern, Bern, Midscale, 2024-04-04, 2024-04-16, FALSE);执行bq query --project_id$GOOGLE_CLOUD_PROJECT --dataset_id$BQ_DATASET_NAME --use_legacy_sqlfalse insert_hotels_data.sql三、Step 2安装并配置 Toolbox这一步将下载 Toolbox 二进制、在tools.yaml中声明指向 BigQuery 的 source 与 tools然后启动 Toolbox Server。3.1 下载 Toolbox 二进制按你的操作系统与 CPU 架构选择对应二进制支持linux/amd64、darwin/arm64、darwin/amd64、windows/amd64、windows/arm64export OSlinux/amd64 # 从上述平台中选择一个 curl -O https://storage.googleapis.com/mcp-toolbox-for-databases/v0.30.0/$OS/toolbox赋予执行权限chmod x toolbox3.2 编写 tools.yaml将下面的内容写入tools.yaml。你需要把其中的YOUR_PROJECT_ID与YOUR_DATASET_NAME占位符替换为真实的 BigQuery 项目与数据集名location字段可选缺省为us。hotels表名直接在 statement 中使用。kind: source name: my-bigquery-source type: bigquery project: YOUR_PROJECT_ID location: us --- kind: tool name: search-hotels-by-name type: bigquery-sql source: my-bigquery-source description: Search for hotels based on name. parameters: - name: name type: string description: The name of the hotel. statement: SELECT * FROM YOUR_DATASET_NAME.hotels WHERE LOWER(name) LIKE LOWER(CONCAT(%, name, %)); --- kind: tool name: search-hotels-by-location type: bigquery-sql source: my-bigquery-source description: Search for hotels based on location. parameters: - name: location type: string description: The location of the hotel. statement: SELECT * FROM YOUR_DATASET_NAME.hotels WHERE LOWER(location) LIKE LOWER(CONCAT(%, location, %)); --- kind: tool name: book-hotel type: bigquery-sql source: my-bigquery-source description: - Book a hotel by its ID. If the hotel is successfully booked, returns a NULL, raises an error if not. parameters: - name: hotel_id type: integer description: The ID of the hotel to book. statement: UPDATE YOUR_DATASET_NAME.hotels SET booked TRUE WHERE id hotel_id; --- kind: tool name: update-hotel type: bigquery-sql source: my-bigquery-source description: - Update a hotels check-in and check-out dates by its ID. Returns a message indicating whether the hotel was successfully updated or not. parameters: - name: checkin_date type: string description: The new check-in date of the hotel. - name: checkout_date type: string description: The new check-out date of the hotel. - name: hotel_id type: integer description: The ID of the hotel to update. statement: - UPDATE YOUR_DATASET_NAME.hotels SET checkin_date PARSE_DATE(%Y-%m-%d, checkin_date), checkout_date PARSE_DATE(%Y-%m-%d, checkout_date) WHERE id hotel_id; --- kind: tool name: cancel-hotel type: bigquery-sql source: my-bigquery-source description: Cancel a hotel by its ID. parameters: - name: hotel_id type: integer description: The ID of the hotel to cancel. statement: UPDATE YOUR_DATASET_NAME.hotels SET booked FALSE WHERE id hotel_id;关于toolset的重要说明上面的tools.yaml没有包含toolset类型。第三步中的 Python 示例例如await toolbox_client.load_toolset(my-toolset)依赖一个名为my-toolset的 toolset。要让这些示例直接工作你需要在tools.yaml中补充 toolset 定义# 使用 load_toolset(my-toolset) 时请将此段追加到 tools.yaml kind: toolset name: my-toolset tools: - search-hotels-by-name - search-hotels-by-location - book-hotel - update-hotel - cancel-hotel当然你也可以修改 Agent 代码逐个加载工具例如使用await toolbox_client.load_tool(search-hotels-by-name)。3.3 启动 Toolbox Server./toolbox --config tools.yaml注意Toolbox 默认开启配置动态重载dynamic reloading如需关闭请使用--disable-reload标志。3.4 深入bigquery 数据源配置项全解析除了示例中的project与locationbigquery数据源还支持一系列可选字段它们在 BigQuery 数据源文档 中有完整参考并在 源码Config结构体 中落地。其中默认值可对照 源码newConfigMaxQueryResultRows默认为 50WriteMode缺省为allowed且设置了readOnly: true时自动降级为blocked字段类型必填说明typestring是固定为bigqueryprojectstring是用于计费及作为 BigQuery 资源默认项目的 GCP 项目 IDlocationstring否查询作业的运行区域如us、asia-northeast1必须与查询所引用表的位置一致无法确定时默认USreadOnlyboolean否控制整个 Toolbox 层是否只读为true时配合 MCP 只读注解与工具抑制并默认将writeMode置为blockedwriteModestring否allowed默认允许所有查询/blocked严格只读仅允许SELECT并抑制注册写能力工具/protected基于共享 BigQuery session 执行允许临时表等有状态操作但保护永久数据集不能与useClientOAuth: true同时使用不适合多用户环境allowedDatasets[]string否允许访问的数据集白名单访问名单外数据集会被拒绝同时禁止CREATE SCHEMA等数据集级操作及无法静态分析表访问的操作useClientOAuthstring否设为true时转发客户端默认Authorization头中的 OAuth token也可指定自定义请求头名空串或false关闭scopes[]string否凭据使用的 OAuth 2.0 作用域列表缺省用默认作用域impersonateServiceAccountstring否调用 BigQuery/Dataplex API 时模拟的服务账号邮箱maxQueryResultRowsint否单次查询返回的最大行数默认 50maximumBytesBilledint64否单次查询的计费字节上限超限在真正执行前即失败apiEndpointstring否覆盖 BigQuery API 端点可用于代理或本地模拟器sqlCommenterboolean否覆盖全局--sql-commenter开关对本数据源单独生效从 bigquery-sql 工具实现 看protected模式下工具会先获取数据源共享的 BigQuery session并把session_id作为连接属性传入从而支持CREATE TEMP TABLE之类的有状态操作。而writeMode: blocked时bigquery-execute-sql等动态工具会拒绝非SELECT语句见 bigqueryexecutesql.go。如果你不想手写完整配置仓库还提供了 BigQuery 预置配置用环境变量占位的方式声明了bigquery-source及execute_sql、list_dataset_ids、get_table_info、search_catalog、analyze_contribution等一组常用工具以及data、analytics两个工具分组可通过${BIGQUERY_PROJECT}、${BIGQUERY_LOCATION:}、${BIGQUERY_MAX_QUERY_RESULT_ROWS:50}等变量快速注入配置。3.5 深入bigquery-sql 工具与参数化查询bigquery-sql执行一条预定义的 GoogleSQL 语句通过parameters声明可插入查询的参数。工具支持命名参数如name与位置参数?但两者不能在同一查询中混用且参数只能作为表达式占位不能替代标识符、列名、表名等查询成分——这正是防止 SQL 注入的关键设计。在 bigquerysql.go 中可以看到工具在执行前会先调用bqutil.DryRunQuery做一次 dry-run 预检这也是maximumBytesBilled能先于真正执行拦截超限查询的原因随后再以参数化方式真正执行语句。参数支持的类型包括string、integer、float、boolean、数组等源码的buildQueryParameters会为 dry-run 与正式执行分别构造低层REST与高层Go client两套参数数组参数还会自动做类型转换与精度处理。除基础parameters外工具还支持templateParameters如SELECT * FROM {{.tableName}}它允许直接修改语句中的标识符与表名但更易受 SQL 注入影响官方建议仅在确有必要时使用详见 bigquery-sql 工具文档 中的示例与参考表格。此外bigquery-sql还支持搭配 embeddingModel 做向量检索将文本参数用embeddedBy标注后工具会自动把文本转成 BigQuery 所需的ARRAYFLOAT64配合ML.DISTANCE实现语义搜索。四、Step 3将 Agent 连接到 Toolbox下面编写并运行一个从 Toolbox 加载工具、由大模型驱动的 Agent。若你想在 Google Colab 中实验可以连接本地运行时。4.1 安装 SDK 包新开一个终端按框架选择# Core pip install toolbox-core # LangChain pip install toolbox-langchain # LlamaIndex pip install toolbox-llamaindex # ADK pip install google-adk[toolbox]4.2 安装其余依赖# Core / LangChain按需换成对应包 pip install langgraph langchain-google-vertexai # pip install langchain-google-genai # pip install langchain-anthropic # LlamaIndex pip install llama-index-llms-google-genai # pip install llama-index-llms-anthropic # ADK无需其他依赖4.3 编写 Agent 代码新建文件hotel_agent.py从下面四种实现中选择一种它们都加载同一个my-toolset并在本地 5000 端口访问 Toolbox Server。Core原生 Gemini function calling使用toolbox_core.ToolboxClient加载工具集把每个工具经FunctionDeclaration.from_callable_with_api_option转成 Gemini 的函数声明再手动完成 function-call 循环import asyncio from google import genai from google.genai.types import ( Content, FunctionDeclaration, GenerateContentConfig, Part, Tool, ) from toolbox_core 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 id 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., Please book the hotel Hilton Basel for me., This is too expensive. Please cancel it., Please book Hyatt Regency for me, My check in dates for my booking would be from April 10, 2024 to April 19, 2024., ] async def run_application(): async with ToolboxClient(http://127.0.0.1:5000) as toolbox_client: toolbox_tools await toolbox_client.load_toolset(my-toolset) tool_map {tool.__name__: tool for tool in toolbox_tools} genai_client genai.Client( vertexaiTrue, projectproject-id, locationus-central1 ) genai_tools [ Tool( function_declarations[ FunctionDeclaration.from_callable_with_api_option(callabletool) ] ) for tool in toolbox_tools ] history [] for query in queries: user_prompt_content Content( roleuser, parts[Part.from_text(textquery)], ) history.append(user_prompt_content) response genai_client.models.generate_content( modelgemini-2.0-flash-001, contentshistory, configGenerateContentConfig( system_instructionprompt, toolsgenai_tools, ), ) history.append(response.candidates[0].content) function_response_parts [] for function_call in response.function_calls: fn_name function_call.name if fn_name in tool_map: function_result await tool_mapfn_name else: raise ValueError(fFunction name {fn_name} not present.) function_response {result: function_result} function_response_part Part.from_function_response( namefunction_call.name, responsefunction_response, ) function_response_parts.append(function_response_part) if function_response_parts: tool_response_content Content(roletool, partsfunction_response_parts) history.append(tool_response_content) response2 genai_client.models.generate_content( modelgemini-2.0-flash-001, contentshistory, configGenerateContentConfig( toolsgenai_tools, ), ) final_model_response_content response2.candidates[0].content history.append(final_model_response_content) print(response2.text) asyncio.run(run_application())LangChain通过toolbox_langchain.ToolboxClient加载工具交给 LangGraph 的create_react_agent编排并使用MemorySaver维护会话线程import asyncio from langgraph.prebuilt import create_react_agent # TODO(developer): replace this with another import if needed from langchain_google_vertexai import ChatVertexAI # 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 ChatVertexAI(model_namegemini-2.0-flash-001) # model ChatGoogleGenerativeAI(modelgemini-2.0-flash-001) # model ChatAnthropic(modelclaude-3-5-sonnet-20240620) # Load the tools from the Toolbox server client ToolboxClient(http://127.0.0.1:5000) 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)]} response await agent.ainvoke(inputs, stream_modevalues, configconfig) print(response[messages][-1].content) asyncio.run(main())LlamaIndex用toolbox_llamaindex.ToolboxClient加载工具交给AgentWorkflow运行import asyncio import os from llama_index.core.agent.workflow import AgentWorkflow from llama_index.core.workflow import Context # TODO(developer): replace this with another import if needed from llama_index.llms.google_genai import GoogleGenAI # from llama_index.llms.anthropic import Anthropic from toolbox_llamaindex 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 llm GoogleGenAI( modelgemini-2.0-flash-001, vertexai_config{location: us-central1}, ) # llm GoogleGenAI( # api_keyos.getenv(GOOGLE_API_KEY), # modelgemini-2.0-flash-001, # ) # llm Anthropic( # modelclaude-3-7-sonnet-latest, # api_keyos.getenv(ANTHROPIC_API_KEY) # ) # Load the tools from the Toolbox server client ToolboxClient(http://127.0.0.1:5000) tools await client.aload_toolset() agent AgentWorkflow.from_tools_or_functions( tools, llmllm, system_promptprompt, ) ctx Context(agent) for query in queries: response await agent.arun(user_msgquery, ctxctx) print(f---- {query} ----) print(str(response)) asyncio.run(main())ADKAgent Development Kit使用ToolboxToolset直接挂载 Toolbox Server 上加载的工具集无需额外依赖from google.adk.agents import Agent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService from google.adk.tools.toolbox_toolset import ToolboxToolset from google.genai import types # For constructing message content import os os.environ[GOOGLE_GENAI_USE_VERTEXAI] True # TODO(developer): Replace YOUR_PROJECT_ID with your Google Cloud Project ID os.environ[GOOGLE_CLOUD_PROJECT] YOUR_PROJECT_ID # TODO(developer): Replace us-central1 with your Google Cloud Location (region) os.environ[GOOGLE_CLOUD_LOCATION] us-central1 # --- Load Tools from Toolbox --- # TODO(developer): Ensure the Toolbox server is running at http://127.0.0.1:5000 toolset ToolboxToolset(server_urlhttp://127.0.0.1:5000) # --- Define the Agents Prompt --- 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. # --- Configure the Agent --- root_agent Agent( modelgemini-2.0-flash-001, namehotel_agent, descriptionA helpful AI assistant that can search and book hotels., instructionprompt, tools[toolset], # Pass the loaded toolset ) # --- Initialize Services for Running the Agent --- session_service InMemorySessionService() artifacts_service InMemoryArtifactService() runner Runner( app_namehotel_agent, agentroot_agent, artifact_serviceartifacts_service, session_servicesession_service, ) async def main(): # Create a new session for the interaction. session await session_service.create_session( state{}, app_namehotel_agent, user_id123 ) # --- Define Queries and Run the Agent --- 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., ] for query in queries: content types.Content(roleuser, parts[types.Part(textquery)]) events runner.run(session_idsession.id, user_id123, new_messagecontent) responses ( part.text for event in events for part in event.content.parts if part.text is not None ) for text in responses: print(text) import asyncio if __name__ __main__: asyncio.run(main())4.4 运行 Agentpython hotel_agent.py观察输出Agent 应能按用户的自然语言依次完成「搜索 Basel 的酒店 → 预订 Hilton Basel → 取消 → 改订 Hyatt Regency → 修改入住/离店日期」等操作每次操作都通过 Toolbox 提供的bigquery-sql工具落到 BigQuery 表上。五、验证思路与进一步阅读验证数据是否真的被修改Agent 执行预订/取消后可以在 BigQuery Console 中查询hotels表的booked、checkin_date、checkout_date字段或用bq query手工确认。观察 Toolbox 行为本地运行期间可查看 Server 日志确认工具注册、参数解析与 SQL 执行链路配置变更后无需重启默认动态重载也可用--disable-reload关闭该特性。更完整的工具集BigQuery 集成不止bigquery-sql还包括bigquery-execute-sql执行任意 SQL、bigquery-get-table-info、bigquery-list-dataset-ids、bigquery-search-catalog、bigquery-forecast、bigquery-analyze-contribution、bigquery-conversational-analytics等完整清单见 BigQuery 工具文档目录。配置进阶readOnly/writeMode与 MCP 只读注解、工具抑制的联动关系可参考 source.mdparameters与templateParameters的完整字段定义见 工具配置文档SQL Commenter 以 job label 形式注入查询可观测性详见 监控文档。生产注意将useClientOAuth用于「代表客户端/最终用户」的查询场景时务必确保对应身份具备所需 IAM 权限allowedDatasets与maximumBytesBilled是控制 Agent 访问面与成本的有效手段。【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 MCP Toolbox 快速搭建 BigQuery 酒店预订智能体：本地端到端上手指南
用 MCP Toolbox 快速搭建 BigQuery 酒店预订智能体本地端到端上手指南【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolboxMCP Toolbox for Databases下称 Toolbox是一个面向数据库的开源 MCP Server可以将 BigQuery 等数据源封装成可供 LLM 直接调用的工具。本文以「酒店搜索 / 预订 / 改期 / 取消」为例完整演示在本地环境从零构建一个可运行的大模型智能体Agent先通过bqCLI 准备 BigQuery 数据集与表再编写tools.yaml将表操作声明为bigquery-sql工具并启动 Toolbox Server最后分别用 Toolbox Core SDK、LangChain、LlamaIndex 与 Google ADK 四种方式把 Agent 接到 Toolbox 上执行真实查询。读完本文你将掌握 Toolbox 与 BigQuery 集成的最小可运行闭环以及bigquery数据源与bigquery-sql工具的配置要点。本文内容以仓库中的 local_quickstart.md 为骨架并结合 BigQuery 数据源文档、bigquery-sql 工具文档、bigquery-sql 实现源码 与 BigQuery 预置配置 进行源码级补充。一、开始之前环境准备本文假设你已经完成以下准备工作安装 Python 3.10并确保pip可用建议使用venv等虚拟环境工具管理依赖。安装并配置 Google Cloud SDKgcloudCLI。通过 ADCApplication Default Credentials完成身份认证gcloud auth login --update-adc设置默认 GCP 项目将YOUR_PROJECT_ID替换为你的真实项目 IDgcloud config set project YOUR_PROJECT_ID export GOOGLE_CLOUD_PROJECTYOUR_PROJECT_IDToolbox 与各客户端库会默认使用该项目操作 BigQuery除非在配置中被显式覆盖。在 GCP 项目中启用 BigQuery API。安装 BigQuery 的 Python 客户端库pip install google-cloud-bigquery按所选框架完成 LLM 接入准备Core / LangChain安装langchain-vertexai、langchain-google-genai或langchain-anthropic中的对应包LlamaIndex安装llama-index-llms-google-genai或llama-index-llms-anthropicADK安装google-adk包。从源码看Toolbox 的 BigQuery 集成默认依赖 Application Default Credentials 完成认证。相关字段如scopes、useClientOAuth、impersonateServiceAccount在 BigQuery 数据源实现 的Config结构体中有明确定义稍后会在配置章节详解。二、Step 1在 BigQuery 中创建数据集与表先为 Agent 准备一份需要访问的测试数据。所有 BigQuery 操作都基于你在前面配置好的 Google Cloud 项目执行。2.1 创建数据集export BQ_DATASET_NAMEYOUR_DATASET_NAME # 例如 toolbox_ds export BQ_LOCATIONUS # 例如 US、EU、asia-northeast1 bq --location$BQ_LOCATION mk $BQ_DATASET_NAME也可以通过 Google Cloud Console 的 BigQuery 页面完成同样的操作。权限提示在生产环境中请确保运行 Toolbox 的服务账号或用户对数据集/项目具备必要的 IAM 权限例如 BigQuery Data Editor、BigQuery User。本地快速开始使用用户凭据你的自有权限即生效。2.2 创建 hotels 表新建文件create_hotels_table.sql内容如下CREATE TABLE IF NOT EXISTS YOUR_PROJECT_ID.YOUR_DATASET_NAME.hotels ( id INT64 NOT NULL, name STRING NOT NULL, location STRING NOT NULL, price_tier STRING NOT NULL, checkin_date DATE NOT NULL, checkout_date DATE NOT NULL, booked BOOLEAN NOT NULL );注意请把 SQL 中的YOUR_PROJECT_ID与YOUR_DATASET_NAME替换成真实值。然后执行该 SQLbq query --project_id$GOOGLE_CLOUD_PROJECT --dataset_id$BQ_DATASET_NAME --use_legacy_sqlfalse create_hotels_table.sql2.3 插入初始数据新建文件insert_hotels_data.sql加入如下 INSERT 语句同样替换项目与数据集占位符INSERT INTO YOUR_PROJECT_ID.YOUR_DATASET_NAME.hotels (id, name, location, price_tier, checkin_date, checkout_date, booked) VALUES (1, Hilton Basel, Basel, Luxury, 2024-04-20, 2024-04-22, FALSE), (2, Marriott Zurich, Zurich, Upscale, 2024-04-14, 2024-04-21, FALSE), (3, Hyatt Regency Basel, Basel, Upper Upscale, 2024-04-02, 2024-04-20, FALSE), (4, Radisson Blu Lucerne, Lucerne, Midscale, 2024-04-05, 2024-04-24, FALSE), (5, Best Western Bern, Bern, Upper Midscale, 2024-04-01, 2024-04-23, FALSE), (6, InterContinental Geneva, Geneva, Luxury, 2024-04-23, 2024-04-28, FALSE), (7, Sheraton Zurich, Zurich, Upper Upscale, 2024-04-02, 2024-04-27, FALSE), (8, Holiday Inn Basel, Basel, Upper Midscale, 2024-04-09, 2024-04-24, FALSE), (9, Courtyard Zurich, Zurich, Upscale, 2024-04-03, 2024-04-13, FALSE), (10, Comfort Inn Bern, Bern, Midscale, 2024-04-04, 2024-04-16, FALSE);执行bq query --project_id$GOOGLE_CLOUD_PROJECT --dataset_id$BQ_DATASET_NAME --use_legacy_sqlfalse insert_hotels_data.sql三、Step 2安装并配置 Toolbox这一步将下载 Toolbox 二进制、在tools.yaml中声明指向 BigQuery 的 source 与 tools然后启动 Toolbox Server。3.1 下载 Toolbox 二进制按你的操作系统与 CPU 架构选择对应二进制支持linux/amd64、darwin/arm64、darwin/amd64、windows/amd64、windows/arm64export OSlinux/amd64 # 从上述平台中选择一个 curl -O https://storage.googleapis.com/mcp-toolbox-for-databases/v0.30.0/$OS/toolbox赋予执行权限chmod x toolbox3.2 编写 tools.yaml将下面的内容写入tools.yaml。你需要把其中的YOUR_PROJECT_ID与YOUR_DATASET_NAME占位符替换为真实的 BigQuery 项目与数据集名location字段可选缺省为us。hotels表名直接在 statement 中使用。kind: source name: my-bigquery-source type: bigquery project: YOUR_PROJECT_ID location: us --- kind: tool name: search-hotels-by-name type: bigquery-sql source: my-bigquery-source description: Search for hotels based on name. parameters: - name: name type: string description: The name of the hotel. statement: SELECT * FROM YOUR_DATASET_NAME.hotels WHERE LOWER(name) LIKE LOWER(CONCAT(%, name, %)); --- kind: tool name: search-hotels-by-location type: bigquery-sql source: my-bigquery-source description: Search for hotels based on location. parameters: - name: location type: string description: The location of the hotel. statement: SELECT * FROM YOUR_DATASET_NAME.hotels WHERE LOWER(location) LIKE LOWER(CONCAT(%, location, %)); --- kind: tool name: book-hotel type: bigquery-sql source: my-bigquery-source description: - Book a hotel by its ID. If the hotel is successfully booked, returns a NULL, raises an error if not. parameters: - name: hotel_id type: integer description: The ID of the hotel to book. statement: UPDATE YOUR_DATASET_NAME.hotels SET booked TRUE WHERE id hotel_id; --- kind: tool name: update-hotel type: bigquery-sql source: my-bigquery-source description: - Update a hotels check-in and check-out dates by its ID. Returns a message indicating whether the hotel was successfully updated or not. parameters: - name: checkin_date type: string description: The new check-in date of the hotel. - name: checkout_date type: string description: The new check-out date of the hotel. - name: hotel_id type: integer description: The ID of the hotel to update. statement: - UPDATE YOUR_DATASET_NAME.hotels SET checkin_date PARSE_DATE(%Y-%m-%d, checkin_date), checkout_date PARSE_DATE(%Y-%m-%d, checkout_date) WHERE id hotel_id; --- kind: tool name: cancel-hotel type: bigquery-sql source: my-bigquery-source description: Cancel a hotel by its ID. parameters: - name: hotel_id type: integer description: The ID of the hotel to cancel. statement: UPDATE YOUR_DATASET_NAME.hotels SET booked FALSE WHERE id hotel_id;关于toolset的重要说明上面的tools.yaml没有包含toolset类型。第三步中的 Python 示例例如await toolbox_client.load_toolset(my-toolset)依赖一个名为my-toolset的 toolset。要让这些示例直接工作你需要在tools.yaml中补充 toolset 定义# 使用 load_toolset(my-toolset) 时请将此段追加到 tools.yaml kind: toolset name: my-toolset tools: - search-hotels-by-name - search-hotels-by-location - book-hotel - update-hotel - cancel-hotel当然你也可以修改 Agent 代码逐个加载工具例如使用await toolbox_client.load_tool(search-hotels-by-name)。3.3 启动 Toolbox Server./toolbox --config tools.yaml注意Toolbox 默认开启配置动态重载dynamic reloading如需关闭请使用--disable-reload标志。3.4 深入bigquery 数据源配置项全解析除了示例中的project与locationbigquery数据源还支持一系列可选字段它们在 BigQuery 数据源文档 中有完整参考并在 源码Config结构体 中落地。其中默认值可对照 源码newConfigMaxQueryResultRows默认为 50WriteMode缺省为allowed且设置了readOnly: true时自动降级为blocked字段类型必填说明typestring是固定为bigqueryprojectstring是用于计费及作为 BigQuery 资源默认项目的 GCP 项目 IDlocationstring否查询作业的运行区域如us、asia-northeast1必须与查询所引用表的位置一致无法确定时默认USreadOnlyboolean否控制整个 Toolbox 层是否只读为true时配合 MCP 只读注解与工具抑制并默认将writeMode置为blockedwriteModestring否allowed默认允许所有查询/blocked严格只读仅允许SELECT并抑制注册写能力工具/protected基于共享 BigQuery session 执行允许临时表等有状态操作但保护永久数据集不能与useClientOAuth: true同时使用不适合多用户环境allowedDatasets[]string否允许访问的数据集白名单访问名单外数据集会被拒绝同时禁止CREATE SCHEMA等数据集级操作及无法静态分析表访问的操作useClientOAuthstring否设为true时转发客户端默认Authorization头中的 OAuth token也可指定自定义请求头名空串或false关闭scopes[]string否凭据使用的 OAuth 2.0 作用域列表缺省用默认作用域impersonateServiceAccountstring否调用 BigQuery/Dataplex API 时模拟的服务账号邮箱maxQueryResultRowsint否单次查询返回的最大行数默认 50maximumBytesBilledint64否单次查询的计费字节上限超限在真正执行前即失败apiEndpointstring否覆盖 BigQuery API 端点可用于代理或本地模拟器sqlCommenterboolean否覆盖全局--sql-commenter开关对本数据源单独生效从 bigquery-sql 工具实现 看protected模式下工具会先获取数据源共享的 BigQuery session并把session_id作为连接属性传入从而支持CREATE TEMP TABLE之类的有状态操作。而writeMode: blocked时bigquery-execute-sql等动态工具会拒绝非SELECT语句见 bigqueryexecutesql.go。如果你不想手写完整配置仓库还提供了 BigQuery 预置配置用环境变量占位的方式声明了bigquery-source及execute_sql、list_dataset_ids、get_table_info、search_catalog、analyze_contribution等一组常用工具以及data、analytics两个工具分组可通过${BIGQUERY_PROJECT}、${BIGQUERY_LOCATION:}、${BIGQUERY_MAX_QUERY_RESULT_ROWS:50}等变量快速注入配置。3.5 深入bigquery-sql 工具与参数化查询bigquery-sql执行一条预定义的 GoogleSQL 语句通过parameters声明可插入查询的参数。工具支持命名参数如name与位置参数?但两者不能在同一查询中混用且参数只能作为表达式占位不能替代标识符、列名、表名等查询成分——这正是防止 SQL 注入的关键设计。在 bigquerysql.go 中可以看到工具在执行前会先调用bqutil.DryRunQuery做一次 dry-run 预检这也是maximumBytesBilled能先于真正执行拦截超限查询的原因随后再以参数化方式真正执行语句。参数支持的类型包括string、integer、float、boolean、数组等源码的buildQueryParameters会为 dry-run 与正式执行分别构造低层REST与高层Go client两套参数数组参数还会自动做类型转换与精度处理。除基础parameters外工具还支持templateParameters如SELECT * FROM {{.tableName}}它允许直接修改语句中的标识符与表名但更易受 SQL 注入影响官方建议仅在确有必要时使用详见 bigquery-sql 工具文档 中的示例与参考表格。此外bigquery-sql还支持搭配 embeddingModel 做向量检索将文本参数用embeddedBy标注后工具会自动把文本转成 BigQuery 所需的ARRAYFLOAT64配合ML.DISTANCE实现语义搜索。四、Step 3将 Agent 连接到 Toolbox下面编写并运行一个从 Toolbox 加载工具、由大模型驱动的 Agent。若你想在 Google Colab 中实验可以连接本地运行时。4.1 安装 SDK 包新开一个终端按框架选择# Core pip install toolbox-core # LangChain pip install toolbox-langchain # LlamaIndex pip install toolbox-llamaindex # ADK pip install google-adk[toolbox]4.2 安装其余依赖# Core / LangChain按需换成对应包 pip install langgraph langchain-google-vertexai # pip install langchain-google-genai # pip install langchain-anthropic # LlamaIndex pip install llama-index-llms-google-genai # pip install llama-index-llms-anthropic # ADK无需其他依赖4.3 编写 Agent 代码新建文件hotel_agent.py从下面四种实现中选择一种它们都加载同一个my-toolset并在本地 5000 端口访问 Toolbox Server。Core原生 Gemini function calling使用toolbox_core.ToolboxClient加载工具集把每个工具经FunctionDeclaration.from_callable_with_api_option转成 Gemini 的函数声明再手动完成 function-call 循环import asyncio from google import genai from google.genai.types import ( Content, FunctionDeclaration, GenerateContentConfig, Part, Tool, ) from toolbox_core 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 id 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., Please book the hotel Hilton Basel for me., This is too expensive. Please cancel it., Please book Hyatt Regency for me, My check in dates for my booking would be from April 10, 2024 to April 19, 2024., ] async def run_application(): async with ToolboxClient(http://127.0.0.1:5000) as toolbox_client: toolbox_tools await toolbox_client.load_toolset(my-toolset) tool_map {tool.__name__: tool for tool in toolbox_tools} genai_client genai.Client( vertexaiTrue, projectproject-id, locationus-central1 ) genai_tools [ Tool( function_declarations[ FunctionDeclaration.from_callable_with_api_option(callabletool) ] ) for tool in toolbox_tools ] history [] for query in queries: user_prompt_content Content( roleuser, parts[Part.from_text(textquery)], ) history.append(user_prompt_content) response genai_client.models.generate_content( modelgemini-2.0-flash-001, contentshistory, configGenerateContentConfig( system_instructionprompt, toolsgenai_tools, ), ) history.append(response.candidates[0].content) function_response_parts [] for function_call in response.function_calls: fn_name function_call.name if fn_name in tool_map: function_result await tool_mapfn_name else: raise ValueError(fFunction name {fn_name} not present.) function_response {result: function_result} function_response_part Part.from_function_response( namefunction_call.name, responsefunction_response, ) function_response_parts.append(function_response_part) if function_response_parts: tool_response_content Content(roletool, partsfunction_response_parts) history.append(tool_response_content) response2 genai_client.models.generate_content( modelgemini-2.0-flash-001, contentshistory, configGenerateContentConfig( toolsgenai_tools, ), ) final_model_response_content response2.candidates[0].content history.append(final_model_response_content) print(response2.text) asyncio.run(run_application())LangChain通过toolbox_langchain.ToolboxClient加载工具交给 LangGraph 的create_react_agent编排并使用MemorySaver维护会话线程import asyncio from langgraph.prebuilt import create_react_agent # TODO(developer): replace this with another import if needed from langchain_google_vertexai import ChatVertexAI # 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 ChatVertexAI(model_namegemini-2.0-flash-001) # model ChatGoogleGenerativeAI(modelgemini-2.0-flash-001) # model ChatAnthropic(modelclaude-3-5-sonnet-20240620) # Load the tools from the Toolbox server client ToolboxClient(http://127.0.0.1:5000) 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)]} response await agent.ainvoke(inputs, stream_modevalues, configconfig) print(response[messages][-1].content) asyncio.run(main())LlamaIndex用toolbox_llamaindex.ToolboxClient加载工具交给AgentWorkflow运行import asyncio import os from llama_index.core.agent.workflow import AgentWorkflow from llama_index.core.workflow import Context # TODO(developer): replace this with another import if needed from llama_index.llms.google_genai import GoogleGenAI # from llama_index.llms.anthropic import Anthropic from toolbox_llamaindex 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 llm GoogleGenAI( modelgemini-2.0-flash-001, vertexai_config{location: us-central1}, ) # llm GoogleGenAI( # api_keyos.getenv(GOOGLE_API_KEY), # modelgemini-2.0-flash-001, # ) # llm Anthropic( # modelclaude-3-7-sonnet-latest, # api_keyos.getenv(ANTHROPIC_API_KEY) # ) # Load the tools from the Toolbox server client ToolboxClient(http://127.0.0.1:5000) tools await client.aload_toolset() agent AgentWorkflow.from_tools_or_functions( tools, llmllm, system_promptprompt, ) ctx Context(agent) for query in queries: response await agent.arun(user_msgquery, ctxctx) print(f---- {query} ----) print(str(response)) asyncio.run(main())ADKAgent Development Kit使用ToolboxToolset直接挂载 Toolbox Server 上加载的工具集无需额外依赖from google.adk.agents import Agent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService from google.adk.tools.toolbox_toolset import ToolboxToolset from google.genai import types # For constructing message content import os os.environ[GOOGLE_GENAI_USE_VERTEXAI] True # TODO(developer): Replace YOUR_PROJECT_ID with your Google Cloud Project ID os.environ[GOOGLE_CLOUD_PROJECT] YOUR_PROJECT_ID # TODO(developer): Replace us-central1 with your Google Cloud Location (region) os.environ[GOOGLE_CLOUD_LOCATION] us-central1 # --- Load Tools from Toolbox --- # TODO(developer): Ensure the Toolbox server is running at http://127.0.0.1:5000 toolset ToolboxToolset(server_urlhttp://127.0.0.1:5000) # --- Define the Agents Prompt --- 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. # --- Configure the Agent --- root_agent Agent( modelgemini-2.0-flash-001, namehotel_agent, descriptionA helpful AI assistant that can search and book hotels., instructionprompt, tools[toolset], # Pass the loaded toolset ) # --- Initialize Services for Running the Agent --- session_service InMemorySessionService() artifacts_service InMemoryArtifactService() runner Runner( app_namehotel_agent, agentroot_agent, artifact_serviceartifacts_service, session_servicesession_service, ) async def main(): # Create a new session for the interaction. session await session_service.create_session( state{}, app_namehotel_agent, user_id123 ) # --- Define Queries and Run the Agent --- 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., ] for query in queries: content types.Content(roleuser, parts[types.Part(textquery)]) events runner.run(session_idsession.id, user_id123, new_messagecontent) responses ( part.text for event in events for part in event.content.parts if part.text is not None ) for text in responses: print(text) import asyncio if __name__ __main__: asyncio.run(main())4.4 运行 Agentpython hotel_agent.py观察输出Agent 应能按用户的自然语言依次完成「搜索 Basel 的酒店 → 预订 Hilton Basel → 取消 → 改订 Hyatt Regency → 修改入住/离店日期」等操作每次操作都通过 Toolbox 提供的bigquery-sql工具落到 BigQuery 表上。五、验证思路与进一步阅读验证数据是否真的被修改Agent 执行预订/取消后可以在 BigQuery Console 中查询hotels表的booked、checkin_date、checkout_date字段或用bq query手工确认。观察 Toolbox 行为本地运行期间可查看 Server 日志确认工具注册、参数解析与 SQL 执行链路配置变更后无需重启默认动态重载也可用--disable-reload关闭该特性。更完整的工具集BigQuery 集成不止bigquery-sql还包括bigquery-execute-sql执行任意 SQL、bigquery-get-table-info、bigquery-list-dataset-ids、bigquery-search-catalog、bigquery-forecast、bigquery-analyze-contribution、bigquery-conversational-analytics等完整清单见 BigQuery 工具文档目录。配置进阶readOnly/writeMode与 MCP 只读注解、工具抑制的联动关系可参考 source.mdparameters与templateParameters的完整字段定义见 工具配置文档SQL Commenter 以 job label 形式注入查询可观测性详见 监控文档。生产注意将useClientOAuth用于「代表客户端/最终用户」的查询场景时务必确保对应身份具备所需 IAM 权限allowedDatasets与maximumBytesBilled是控制 Agent 访问面与成本的有效手段。【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考