Google ADK 教程：构建基于 MCP 服务器的物流代理

当下构建复杂 AI 代理的最大障碍不是模型，而是集成。如果您希望一个代理访问十种不同的企业工具，通常就得写十个不同的 API 包装器。我的经验是，这会带来“包装器疲劳”以及巨大的维护负担。每当 API 发生变化，您的代理就会出问题。

解决方案在于 Google Agent Development Kit（ADK） 与 Model Context Protocol（MCP） 的结合。如果把 AI 看作一名员工，ADK 是“头脑”（编排逻辑与记忆），而 MCP 则是“双手”（以标准化方式拾取并使用任意工具）。

在本教程中，我们将使用 ADK 构建一个真实可用的物流补救代理。该代理将：

1. 查询 BigQuery 中高优先级且延迟的订单。
2. 使用 Google 地图 MCP 服务器计算从仓库到客户的实际驾车距离。
3. 给出补救建议，例如对长距离运输升级为空运。

以上全部无需编写任何自定义 API 代码。代理会自动发现并使用工具。

初次接触 Google Cloud？建议先从免费的 Introduction to GCP 课程开始，了解计费、项目与基础服务。

什么是 Google ADK？

Google Agent Development Kit（ADK） 是 Google 推出的开源、代码优先框架，用于构建、测试与部署 AI 代理。与 LangChain 等偏重“链式”理念的框架不同，ADK 专注执行。它为代理的思考方式、工具集使用方法以及在生产环境的运行提供了标准化结构。

核心概念如下：

• Agents（代理）：由其模型（如 Gemini 2.0 Flash）与特定指令集定义的核心实体。
• Toolsets（工具集）：能力的集合。我们无需手动定义工具，而是将 ADK 指向一个 MCP 服务器，代理会自动“学习”这些工具。
• Runners（运行器）：管理会话状态与迭代“思考—行动—观察”循环的执行引擎。

采用代码优先的方法让我们像常规软件一样使用版本控制、测试与 CI/CD。它还提供与 Google 的深度集成，这意味着针对 BigQuery 或 Maps 的托管 MCP 服务器几乎无需部署即可使用。

想了解三大云服务商及其服务的对比，请查看我们的指南：AWS vs. Azure vs. Google Cloud。

Google ADK 的集成架构与逻辑流程

我们的物流补救代理遵循清晰的推理路径来解决业务问题。理解该流程有助于您稍后编写提示指令。

推理循环包括以下步骤：

1. 代理调用 BigQuery MCP，查找 status='Delayed' 且 priority='High' 的行。

2. 对每个返回的订单，代理提取仓库与客户地址。

3. 代理为每个订单调用 Google 地图 MCP 计算驾车距离。

4. 代理应用业务逻辑（例如距离超过 500 英里则建议空运），并以 Markdown 表格形式呈现建议。

设置 MCP 服务器

开始编码之前，我们需要准备底层云基础设施并就绪 MCP 服务器。根据我的经验，正确配置 IAM 权限是此阶段最关键的部分。

步骤 1：初始化 Google Cloud CLI

请先确保已安装 Google Cloud CLI。打开终端，运行以下命令将本地环境与您的 Google 账号关联：

gcloud init技术支持

运行后会自动打开浏览器窗口，提示您使用 Google 帐号登录。登录后，点击 Allow 授权 CLI 管理您的资源。

回到终端，系统会提示您选择默认项目。如果已创建，选择该项目的 ID；也可以在终端中新建项目。

步骤 2：设置计费

要使用 API 服务，您需要设置计费。需要说明的是，Google Cloud 为新用户提供 90 天内 300 美元的免费额度，可用于试用 Compute、GKE、Maps 等服务，这些额度足以支持本教程示例。只有当您启用按量计费或预付费时才会产生费用。设置计费步骤如下：

1. 前往 Google 控制台计费页面，使用您在 gcloud 中使用的账号登录。

2. 如果尚无计费账号，请创建一个（需要添加支付方式，但会优先使用免费额度）。

3. 将计费账号链接到您的项目：Billing → Link a project → 选择项目，或运行：

gcloud billing projects link YOUR_PROJECT_ID --billing-account=BILLING_ACCOUNT_ID技术支持

步骤 3：启用核心 API 与 MCP 服务

接下来需要启用主要的产品 API，以便代理访问 BigQuery 与 Maps 等底层数据服务。

gcloud services enable bigquery.googleapis.com
gcloud services enable compute.googleapis.com
gcloud services enable container.googleapis.com
gcloud services enable mapstools.googleapis.com技术支持

现在需要启用托管的 MCP 端点。可以将其理解为位于常规 API 之上的“翻译器”接口。

首先，将项目 ID 与账号邮箱存入环境变量，避免后续每条命令手动输入：

export PROJECT_ID=$(gcloud config get-value project)
export USER_EMAIL=$(gcloud config get-value account)技术支持

设置好变量后，运行以下命令为您的项目开启 MCP 测试版功能。这些命令会让 Google 为您的代理准备远程 MCP 端点：

gcloud beta services mcp enable bigquery.googleapis.com --project=$PROJECT_ID
gcloud beta services mcp enable mapstools.googleapis.com --project=$PROJECT_ID
gcloud beta services mcp enable container.googleapis.com --project=$PROJECT_ID
gcloud beta services mcp enable compute.googleapis.com --project=$PROJECT_ID技术支持

步骤 3：配置 IAM 与 Maps 密钥

MCP 采用“双层”安全模型。您既需要访问 MCP 服务器本身的权限，也需要单独的数据访问权限（如查看表中的行）。您可以使用 gcloud CLI 或 Google Cloud Console 界面来完成。

IAM 选项 A：使用 gcloud CLI

为您的账号授予 MCP 与 BigQuery 相关权限，运行以下三条命令：

# 与 MCP Translator 通信的权限
gcloud projects add-iam-policy-binding $PROJECT_ID --member="user:$USER_EMAIL" --role="roles/mcp.toolUser"

# 运行 BigQuery 作业与查看数据的权限
gcloud projects add-iam-policy-binding $PROJECT_ID --member="user:$USER_EMAIL" --role="roles/bigquery.jobUser"
gcloud projects add-iam-policy-binding $PROJECT_ID --member="user:$USER_EMAIL" --role="roles/bigquery.dataViewer"技术支持

IAM 选项 B：使用 Google Cloud Console（界面）

如果您更偏好可视化界面，可在浏览器中按以下步骤操作：

1. 前往 Google Cloud Console 的 IAM & Admin 页面。
2. 在列表中找到您的邮箱地址并点击编辑成员（铅笔）图标。
3. 点击 Add Another Role，搜索 MCP Tool User。
4. 重复上述流程，添加 BigQuery Data Viewer 与 BigQuery Job User。
5. 点击 Save。

创建 Maps API 密钥

最后，创建一个 Maps API 密钥。与使用您登录凭据的 BigQuery 不同，Maps MCP 服务器使用该密钥进行计费与配额跟踪：

gcloud alpha services api-keys create --display-name="Maps-MCP-Key"技术支持

步骤 4：使用 MCP Inspector 在本地验证服务器

在编写 Python 代码之前，我们应使用 MCP Inspector 做一次健康检查。我们将使用随 Node.js 提供的 npx 命令。

它的作用是让您无需永久安装即可立即运行某个包（如 inspector）。在终端中运行：

npx @modelcontextprotocol/inspector技术支持

这会在浏览器中启动一个网页界面（通常为 http://localhost:3000）。以下是我在设置列表中的特定选择，以确保我们测试的是 Google 服务器的远程托管版本，而非本地模拟：

1. Transport type：Streamable HTTP。我们选择它是因为它是 Google 托管云服务器使用的生产级协议。

2. URL：输入 https://mapstools.googleapis.com/mcp。这是直达 Google 地图 MCP 的入口。

3. Connection Type：via Proxy。这一选择至关重要。选择“Proxy”后，inspector 会使用您本地的 gcloud 登录自动处理所有复杂的安全握手。

4. Custom Headers：添加 X-Goog-Api-Key，并粘贴您在步骤 3 中生成的密钥。

5. 点击 Connect。

此时您应看到绿色的Connected 状态。点击List Tools 按钮查看服务器提供的所有能力列表。如果列表中出现 compute_routes，则说明您的基础设施配置正确。

想更深入了解如何为您的代理使用 BigQuery、Maps 与其他 Google 服务，请务必查看这篇关于 Google MCP 服务器 的教程。

使用 Google ADK 构建 MCP 服务器代理

基础设施就绪后，我们可以转到本地开发环境。先从整理项目文件开始。

步骤 1：设置项目结构与依赖

首先，为您的项目创建一个文件夹并设置虚拟环境。

mkdir my-agents
cd my-agents
python -m venv .venv
source .venv/bin/activate技术支持

上述最后一条命令适用于 macOS 或 Linux；如果您使用 Windows，请改用 .venv\Scripts\activate。

接下来，安装 google-adk 包及其支持库。

pip install google-adk google-generativeai python-dotenv httpx  google-auth技术支持

您还需要在项目根目录创建一个 .env 文件，存放 PROJECT_ID 与前一步生成的 MAPS_API_KEY。

步骤 2：配置 MCP 工具集

在名为 tools.py 的脚本中，我们不是写“取数”代码，而是编写连接到“会取数的服务器”的代码。

将工具与逻辑分离能大幅提升可调试性。由于我们要与远程 Google Cloud 端点通信，因此使用 StreamableHTTPConnectionParams 处理认证头。

import os
import google.auth
import google.auth.transport.requests
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams 

# --- BigQuery Toolset ---
def get_bigquery_mcp_toolset():   
    credentials, project_id = google.auth.default(
        scopes=["https://www.googleapis.com/auth/bigquery"]
    )
    credentials.refresh(google.auth.transport.requests.Request())
        
    return MCPToolset(
        connection_params=StreamableHTTPConnectionParams(
            url="https://bigquery.googleapis.com/mcp",
            headers={
                "Authorization": f"Bearer {credentials.token}",
                # 'x-goog-user-project' is vital for BigQuery billing
                "x-goog-user-project": project_id,
                "Content-Type": "application/json"
            }
        )
    )

# --- Maps Toolset ---
def get_maps_mcp_toolset():
    maps_api_key = os.getenv('MAPS_API_KEY')
    
    return MCPToolset(
        connection_params=StreamableHTTPConnectionParams(
            url="https://mapstools.googleapis.com/mcp",
            headers={
                "X-Goog-Api-Key": maps_api_key,
                "Content-Type": "application/json"
            }
        )
    )技术支持

步骤 3：定义代理的使命

现在来构建“头脑”。根据我的经验，这里使用某个 Gemini Flash 模型是最佳选择，因为它们在调用工具方面非常迅速。

我们通过 agent.py 文件中的指令为代理提供标准作业流程。请注意，我们使用清晰的四步流水线，避免代理产生混淆。

from .tools import get_maps_mcp_toolset, get_bigquery_mcp_toolset
from google.adk.agents import LlmAgent

# We're giving the agent a very structured 'Instruction' set. 
# Think of this as the SOP (Standard Operating Procedure) for your agent.
root_agent = LlmAgent(
    model='gemini-2.0-flash',
    name='logistics_recovery_agent',
    instruction="""
      You are an expert Retail Operations Recovery Agent. 
      Your job is to find high-priority problems and solve them using the tools provided.

      Step 1: Use 'execute_sql' to find orders where status is 'Delayed' and priority is 'High'.
      Step 2: For every result, pull the warehouse and customer addresses.
      Step 3: Call the Maps 'compute_routes' tool to find the exact drive distance.
      Step 4: Use this logic: If distance > 500 miles, recommend 'Air Shipping'. 
            Otherwise, recommend 'Regional Express Ground'.
    
      Output the final result as a clean Markdown table.
    """,
    # We simply drop our toolsets here. 
    # ADK handles the complexity of deciding which tool to call when.
    tools=[get_maps_mcp_toolset(), get_bigquery_mcp_toolset()]
)技术支持

步骤 4：运行代理循环

要运行代理，我们使用运行器。该组件跟踪会话状态，使代理了解自己已经完成的内容以及接下来要做什么。

from google.adk.runners import Runner
from google.genai import types

async def main():
    # The Runner acts as the host for our agent.
    runner = Runner(app_name="logistics_app", agent=root_agent)

    # Wrap the user input in the standard GenAI Content format
    user_input = "Identify high-priority delayed orders and suggest recovery shipping methods."
    content = types.Content(role="user", parts=[types.Part(text=user_input)])

    print("--- Starting Smart Logistics Agent ---")
    
    # We use run_async to stream the 'events' (Thoughts and Tool Calls)
    async for event in runner.run_async(user_id="admin_user", new_message=content):
        # We only want to print the text part of the response to the console
        if hasattr(event, 'content') and event.content.parts:
            print(event.content.parts[0].text)技术支持

幕后发生了什么？

当您调用 runner.run_async 时，ADK 会启动一个迭代循环：

1. Gemini 思考：“我需要先查看延迟的订单。”

2. ADK 行动：它调用来自 BigQuery MCP 工具集的 execute_sql 工具。

3. 观察：工具返回订单数据。

4. Gemini 思考：“现在我需要这两个订单的距离。”

5. ADK 行动：它调用两次 Maps MCP 工具。

6. 最终响应：Gemini 汇总所有数据，应用“500 英里”逻辑，并格式化为表格。

在 Web 控制台中测试我们的 Google ADK 代理

虽然终端适合查看日志，ADK 还提供了内置 Web 控制台来可视化代理的内部推理。我发现这对调试非常有帮助。

运行 adk web 启动界面。服务器启动后，在浏览器中打开 http://127.0.0.1:8000。您会看到一个简洁的聊天式界面，在其中发送命令：“Identify high-priority delayed orders and suggest recovery shipping methods.”

分析执行追踪

ADK 控制台提供透明的“追踪”，精确展示代理如何与云基础设施交互。

让我们一步步拆解：

1. 工具握手：注意标有 execute_sql 与 compute_routes 的图标。在追踪中，闪电符号表示一次工具调用。代理并非“猜测”，而是实际访问了 BigQuery MCP 服务器，在您的订单表上运行了特定 SQL 查询。

2. 顺序逻辑：追踪显示了清晰的操作顺序。它先通过 BigQuery 获取数据（第一个对勾）。拿到仓库与客户地址后，您会看到多个 compute_routes 调用。代理在延迟订单上进行智能“循环”，从 Maps MCP 服务器获取真实世界的距离数据。

3. 最终综合：追踪底部展示了最终响应。

看看这些数据：

• ORD-101：距离为 380 英里。
• ORD-103：距离为 352 英里。

由于我们给出的逻辑是：“如果距离 > 500 英里，推荐空运；否则推荐区域特快陆运”，代理正确地将该业务规则应用到两个订单，并将其优雅地格式化为 Markdown 表格。

结语

我们刚刚在没有为 BigQuery 或 Google 地图编写任何自定义 API 代码的情况下，构建了一个可用的物流代理。借助 ADK 与 MCP，我们打造了模块化、安全且面向开发者的架构。

AI 的未来不仅是会“聊天”的机器人，更是能够“行动”的代理。有了这些工具，我们终于朝着这一现实迈进。

本教程可作为您探索 Google ADK 与 MCP 服务器的绝佳起点：

• 将 BigQuery 替换为其他数据集（甚至是本地 SQLite MCP 服务器），体验更换数据源的便捷。
• 将 Maps 替换为其他 MCP 服务器（例如用于集群健康检查的 GKE），验证其即插即用特性。
• 在提示中添加更多业务逻辑（成本对比、邮件草稿），并在 ADK Web 界面实时测试。

如果您想进一步学习代理编排，强烈推荐报名我们的 Associate AI Engineer for Developers 职业路径。