在 Cloud Run 上构建和部署宠物护照代理

1. 概览

在此代码实验中,您将学习如何部署 Pet Passport 应用,这是一款使用 Model Context Protocol (MCP) 将数据分析和位置信息服务相结合的 AI 智能体。

这款应用可根据纽约市的犬种受欢迎程度,帮助用户规划与爱犬共度的完美一天。智能体使用“宏观到微观”的推理链:

  1. 战略性发现 (BigQuery):确定纽约市人口最多的特定犬种邮政编码。
  2. 本地执行(Google 地图):使用该邮政编码作为位置偏差来查找“宠物友好型咖啡馆”和“狗公园”。
  3. 行程生成:将数据组合在一起,以创建包含可点击链接和图片的“宠物护照”行程。

该智能体基于 google-adk 框架构建,并由 Gemini 提供支持。

注意:GitHub 上提供了完整的项目代码,包括前端界面。在本 Codelab 中,我们将重点介绍核心代理逻辑和基础设施设置。

2. 设置和要求

首先,我们来确保开发环境设置正确。

1. 向 Google Cloud 进行身份验证

设置活跃的 Google Cloud 项目并进行身份验证。这是代理访问 BigQuery 和其他服务所必需的。

gcloud config set project [YOUR-PROJECT-ID]
gcloud auth application-default login --project [YOUR-PROJECT-ID]

注意:如果您在身份验证期间遇到有关其他项目的错误,可以停用配额项目并手动设置,从而绕过该错误

gcloud auth application-default login --disable-quota-project
gcloud auth application-default set-quota-project [YOUR-PROJECT-ID]

2. 软件要求

您需要在本地机器上安装以下软件:

  • Python(需要 3.13 版或更高版本)
  • Git(用于下载代码库)

下载代码库

此项目的代码位于 Google MCP 代码库中。克隆代码库并前往项目文件夹:

git clone https://github.com/google/mcp.git
cd examples/petpassport

3. 安装

现在您已经拥有这些文件,接下来我们来设置 Python 环境。

  1. 创建虚拟环境:这样可以隔离依赖项。
    python3 -m venv .venv
    
  2. 激活虚拟环境
    • 在 Linux/macOS 上:
      source .venv/bin/activate
      
    • 在 Windows 上:
      .venv\Scripts\activate
      
  3. 安装依赖项
    pip install google-adk==1.28.0 python-dotenv google-genai pillow uvicorn
    

启用 Cloud API

在项目中启用以下 API:

gcloud services enable \
  bigquery.googleapis.com \
  aiplatform.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com \
  run.googleapis.com \
  storage.googleapis.com

选择区域

在 shell 中将区域设置为环境变量:

export REGION=us-central1

4. 获取 API 密钥

如需使用 Maps 和 Gemini 服务,您需要获取 API 密钥并将其存储在项目根目录中的 .env 文件中。

1. Google 地图 API 密钥

  1. 前往 Google Cloud 控制台
  2. 前往 API 和服务 > 凭据
  3. 依次点击创建凭据 > API 密钥
  4. 复制生成的密钥,并将其作为 MAPS_API_KEY=[YOUR_KEY] 添加到 .env 文件中。
  5. (推荐)限制密钥,使其仅允许 MCP 服务器使用的 Maps API。

2. Gemini API 密钥 (AI Studio)

  1. 前往 Google AI Studio
  2. 点击获取 API 密钥,或前往“API 密钥”部分。
  3. 点击创建 API 密钥
  4. 复制该密钥并将其作为 GEMINI_API_KEY=[YOUR_KEY] 添加到 .env 文件中。

5. 安装依赖项

petpassport/ 文件夹中创建 requirements.txt 文件:

google-adk==1.28.0
python-dotenv
google-genai
pillow

6. 对 MCP 服务器进行身份验证

此应用依赖于 Model Context Protocol (MCP) 服务器来与 Google 地图和 BigQuery 进行交互。如需对这些服务器进行身份验证,您需要配置相应的环境变量和标头。

  1. Google 地图 MCP:需要在 X-Goog-Api-Key 标头中传递有效的地图 API 密钥。
  2. BigQuery MCP:需要具有 BigQuery 服务访问权限的 OAuth 凭据。当在 Cloud Run 上运行时,代理会使用默认的计算服务账号;当在本地运行时,代理会使用本地凭据。

我们在代码库中提供了一个设置脚本 setup/setup_env.sh,可帮助您在 .env 文件中配置这些变量。

7. 创建 BigQuery 表

在代理查询狗牌数据之前,我们需要在 BigQuery 中创建数据集和表,并加载数据。

我们提供了一个设置脚本 setup/setup_bigquery.sh,该脚本会执行以下步骤:

  1. 创建一个名为 pet-passport-data-[PROJECT_ID] 的 Cloud Storage 存储分区来存储原始数据。
  2. 下载纽约市公开的狗许可数据集 (CSV)。
  3. 将 CSV 上传到存储分区。
  4. 创建一个名为 nyc_dogs 的 BigQuery 数据集。
  5. 将相应存储分区中的数据加载到数据集中的名为 licenses 的表中。

如需运行设置脚本,请在终端中执行以下命令:

bash setup/setup_bigquery.sh

8. 连接到 MCP 服务器

此应用的一个关键部分是使用 MCP 连接到数据和服务。在本部分中,您将在名为 petpassport/tools.py 的文件中为 BigQuery 和 Google 地图配置 MCP 工具集。

完成 tools.py 代码

以下是 tools.py 的完整实现,包括 MCP 工具集和用于图片和数据持久性的自定义工具。我们已优化此代码,将存储分区解析移至模块级,从而减少冗余:

import os
import dotenv
import google.auth
import time
import datetime
from google.cloud import storage
from PIL import Image
from google import genai
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams 

MAPS_MCP_URL = "https://mapstools.googleapis.com/mcp" 
BIGQUERY_MCP_URL = "https://bigquery.googleapis.com/mcp" 

PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT', 'project_not_set')
BUCKET_NAME = f"pet-passport-data-{PROJECT_ID}" 

def get_maps_mcp_toolset():
    dotenv.load_dotenv()
    maps_api_key = os.getenv('MAPS_API_KEY', 'no_api_found')
    
    tools = MCPToolset(
        connection_params=StreamableHTTPConnectionParams(
            url=MAPS_MCP_URL,
            headers={    
                "X-Goog-Api-Key": maps_api_key
            },
            timeout=30.0,          
            sse_read_timeout=300.0
        )
    )
    print("Maps MCP Toolset configured.")
    return tools


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())
    oauth_token = credentials.token
        
    HEADERS_WITH_OAUTH = {
        "Authorization": f"Bearer {oauth_token}",
        "x-goog-user-project": project_id
    }

    tools = MCPToolset(
        connection_params=StreamableHTTPConnectionParams(
            url=BIGQUERY_MCP_URL,
            headers=HEADERS_WITH_OAUTH,
            timeout=30.0,          
            sse_read_timeout=300.0
        )
    )
    print("BigQuery MCP Toolset configured.")
    return tools

def generate_pet_passport_photo(prompt: str, image_path: str = None) -> str:
    """Generates an image using gemini-3.1-flash-image-preview based on a prompt and a reference image."""
    client = genai.Client()
    output_path = f"/tmp/pet_passport_{int(time.time())}.png"
    
    try:
        image = Image.open(image_path)
        response = client.models.generate_content(
            model="gemini-3.1-flash-image-preview",
            contents=[prompt, image],
        )
        
        for part in response.parts:
            if part.inline_data is not None:
                generated_image = part.as_image()
                generated_image.save(output_path)
                
                # Upload to GCS and generate signed URL
                try:
                    storage_client = storage.Client()
                    bucket = storage_client.bucket(BUCKET_NAME)
                    blob_name = os.path.basename(output_path)
                    blob = bucket.blob(blob_name)
                    
                    blob.upload_from_filename(output_path)
                    
                    url = blob.generate_signed_url(
                        version="v4",
                        expiration=datetime.timedelta(hours=24),
                        method="GET",
                    )
                    return url
                except Exception as e:
                    print(f"Error uploading image to GCS: {e}")
                    return output_path
                
        raise ValueError("No image was returned by the model.")
    except Exception as e:
        print(f"Error generating image: {e}")
        raise

def save_pet_passport(user_id: str, breed: str, postal_code: str, route_details: str, image_paths: list[str] = None) -> str:
    """Appends the generated itinerary to the user's history in GCS."""
    try:
        storage_client = storage.Client()
        bucket = storage_client.bucket(BUCKET_NAME)
        blob = bucket.blob(f"user-{user_id}.json")
        
        # Download existing or start fresh
        # ... (Implementation details hidden for brevity) ...
        return "Success"
    except Exception as e:
        print(f"Error saving path: {e}")
        raise

代码说明:tools.py

  • get_maps_mcp_toolsetget_bigquery_mcp_toolset 使用正确的端点和身份验证标头配置 MCP 客户端。
  • generate_pet_passport_photo 使用 Gemini 创建场景,并将结果上传到 Google Cloud Storage,然后向前端返回一个签名网址,以便在服务器重启后继续使用。

9. 创建代理

配置好工具后,现在就可以构建智能体的“大脑”了。您将使用智能体开发套件 (ADK) 在名为 petpassport/agent.py 的文件中创建智能体。

完成 agent.py 代码

以下是 agent.py 的完整实现,其中定义了代理及其指令:

import os
import dotenv
import tools
from google.adk.agents import LlmAgent

dotenv.load_dotenv()

PROJECT_ID = os.getenv('GOOGLE_CLOUD_PROJECT', 'project_not_set')

maps_toolset = tools.get_maps_mcp_toolset()
bigquery_toolset = tools.get_bigquery_mcp_toolset()

root_agent = LlmAgent(
    model='gemini-2.5-pro',
    name='root_agent',
    instruction=f"""
        You are the Pet Passport Agent. Your goal is to help users find a fun walking route for their dog in NYC.
        
        When given a breed and a postal code, follow this flow:
        1. **Strategic Discovery:** Use BigQuery to find the most popular neighborhood for that breed in NYC.
        2. **Local Execution:** Use Maps to build a walking route with specific places (parks, cafes) in that area.
        
        **NO DIRECTIONS LINKS:** You must NOT include a Google Maps directions link (e.g., `https://www.google.com/maps/dir/...`) in your final response. Only provide links to individual places.
        
        After generating the itinerary, you MUST call the `save_pet_passport` tool to save this path to the user's profile. Pass a clean summary of the itinerary as `route_details`. The summary should include details (like rating, description from maps).
    """,
    tools=[maps_toolset, bigquery_toolset, tools.generate_pet_passport_photo, tools.save_pet_passport]
)

代码说明:agent.py

  • 我们直接导入 tools(扁平化结构),以支持容器环境。
  • 代理使用 gemini-2.5-pro 进行初始化。
  • 这些指令定义了严格的多步推理链(先是 BigQuery,然后是 Maps),并严格禁止出现或渲染会导致杂乱的步行路线。

10. 在本地运行应用

在部署到 Cloud Run 之前,最好先在本地测试应用。

  1. 确保您位于项目目录中
    cd examples/petpassport
    
  2. 启动 FastAPI 服务器:我们使用 uvicorn 运行应用。入口点是 petpassport 文件夹中的 main.py
    uvicorn petpassport.main:app --reload
    
  3. 打开界面:在浏览器中前往 http://127.0.0.1:8000/ui/,与宠物护照界面互动。

11. 部署到 Cloud Run

智能体已准备就绪,现在可以将其部署到 Cloud Run 了。我们直接使用标准 gcloud 命令来严格控制容器环境。

从项目目录中,运行以下命令:

gcloud run deploy petpassport \
  --source petpassport \
  --region $REGION \
  --allow-unauthenticated \
  --labels dev-tutorial=google-mcp

配置环境变量

部署完成后,在 Google Cloud 控制台中前往 Cloud Run 服务,然后在变量和 Secret 标签页下设置以下环境变量:

  • MAPS_API_KEY:您的 Google 地图 API 密钥。
  • GOOGLE_CLOUD_PROJECT:您的项目 ID。
  • PROJECT_ID:您的项目 ID(旧版模块支持冗余)。

12. 提示示例

尝试使用以下提示与已部署的代理互动:

  1. 标准:“我想在纽约市 10021 附近带我的金毛猎犬去散步。为我们找一条有咖啡馆的路线。”
  2. 不同品种:“我家养的是法国斗牛犬,我们住在上西区(靠近 10024)。建议一条短途步行路线,其中包含一个热门的遛狗公园。”
  3. 使用图片:(上传一张狗狗的照片)“这是我家柯基犬的照片!我们位于 10013 附近。为我们规划完美的一天。”

13. 清理

为避免因本教程中使用的资源产生费用,请执行以下操作:

  • 删除 Cloud Run 服务:gcloud run services delete petpassport --region=$REGION
  • 删除 GCS 存储分区:gcloud storage rm -r gs://pet-passport-data-$PROJECT_ID