Securely connecting Gemini Enterprise to Snowflake data

通过 Google ADK 连接 Gemini 企业版与 Snowflake | Google Cloud 博客

数据分析

安全连接 Gemini 企业版与 Snowflake 数据

2026年10月28日

##### Pankaj Bajpai

云数据工程师

##### Abhilash Adavi

AI 工程师

##### 立即体验 Gemini

进入职场 AI 的大门

立即体验

企业级 AI 需要超越简单的聊天界面。为了提供真正的价值，AI 代理必须访问您的数据仓库以生成可操作的见解，同时始终保持严格的安全性和数据治理。

本指南探讨了如何在 Google Cloud 的 Gemini 企业版与 Snowflake 之间构建安全的桥梁。通过使用模型上下文协议（MCP）和 Google 代理开发工具包（ADK），您可以使代理作为能够理解自然语言的助手，通过 Snowflake Cortex 生成精确的 SQL，并使用用户级别的权限执行查询。

选择合适的集成模式

直接连接 vs. 自定义编排

本文展示的 Google ADK 架构代表了高级/自定义编排模式。它专为在 Google Cloud 上构建复杂多代理系统的开发人员设计，例如结合 Snowflake 洞察与外部企业工具、自定义推理循环或 Gemini Agent Engine 中跨平台状态管理的工作流。

对于标准企业基础用例，若需直接连接 Gemini 企业版与 Snowflake，请使用此链接中提到的联合解决方案。

架构方法

该架构允许业务用户提出类似“上个季度德克萨斯州销量前五的商品是什么？”的问题。

而不是向客户端暴露原始 SQL 执行，代理通过 Snowflake MCP 服务器将查询生成和执行委托给受管控的 Snowflake Cortex 代理：

• 受管控的翻译：ADK 代理将用户的自然语言问题传递给 Snowflake MCP 服务器，该服务器调用由 Cortex Analyst 语义视图支持的 Cortex 代理（CORTEX_AGENT_RUN）。

• 在受管控边界内直接执行：Cortex 代理将问题转换为 SQL，并使用经过身份验证用户的 OAuth 会话和角色权限在 Snowflake 内直接执行，将结构化结果返回给 ADK 代理。

这种方法提供了映射到您企业语义定义的确定性文本到 SQL 转换，将查询执行保留在 Snowflake 的受管控边界内，并减少了模式幻觉或未经验证的 SQL 执行风险。

系统架构流程

下图说明了 Gemini 企业版应用与 Snowflake 生态系统之间的信任边界和编号执行流程。

详细执行顺序

为了突出代理如何通过 MCP 将自然语言查询翻译和受管控执行委托给 Snowflake Cortex 代理，请参见下图的序列图：

步骤 1：设置 Snowflake 安全集成

在企业 AI 中，主要风险是通过通用服务账户授予模型广泛访问权限，这可能导致用户绕过安全限制。我们通过使用 OAuth 2.0 进行用户模拟来解决此问题。

首先，在 Snowflake 中建立一个保密的 OAuth 客户端以保护客户端密钥。以下代码片段在 Snowflake 内配置安全集成：

Loading...

使用 ACCOUNTADMIN 角色；创建或替换安全集成 <SECURITY_INTEGRATION_NAME> 类型 = OAUTH 启用 = TRUE OAUTH_CLIENT = CUSTOM OAUTH_CLIENT_TYPE = 'CONFIDENTIAL' -- Google Cloud 官方握手接收地址：OAUTH_REDIRECT_URI = 'https://vertexaisearch.cloud.google.com/oauth-redirect' OAUTH_ISSUE_REFRESH_TOKENS = TRUE -- 使连接保持活跃 90 天（Snowflake 允许的最大限制）OAUTH_REFRESH_TOKEN_VALIDITY = 7776000；

要获取 Google Cloud 配置所需的生成客户端凭证，请在 Snowflake 中运行以下命令：

-- 获取 OAuth 客户端 ID 和客户端密钥 SELECT SYSTEM$SHOW_OAUTH_CLIENT_SECRETS('<SECURITY_INTEGRATION_NAME>');

第 2 步：数据治理和访问控制

通过在 Snowflake 中为代理创建专用角色来遵循最小权限原则。

USE ROLE SECURITYADMIN；-- 1. 创建专用角色 CREATE ROLE IF NOT EXISTS <GENAI_ROLE_NAME>；GRANT ROLE <GENAI_ROLE_NAME> TO USER <BUSINESS_USER_NAME>；USE ROLE SYSADMIN；-- 2. 授予结构访问权限 GRANT USAGE ON DATABASE <DATABASE_NAME> TO ROLE <GENAI_ROLE_NAME>；GRANT USAGE ON SCHEMA <DATABASE_NAME>.<SCHEMA_NAME> TO ROLE <GENAI_ROLE_NAME>；-- 3. 授予计算访问权限 GRANT USAGE ON WAREHOUSE <WAREHOUSE_NAME> TO ROLE <GENAI_ROLE_NAME>；-- 4. 授予语义层访问权限 GRANT SELECT, REFERENCES ON SEMANTIC VIEW <DATABASE_NAME>.<SCHEMA_NAME>.<SEMANTIC_VIEW_NAME> TO ROLE <GENAI_ROLE_NAME>；-- 5. 授予底层原始数据表的查询权限 GRANT SELECT ON ALL TABLES IN SCHEMA <DATABASE_NAME>.<SCHEMA_NAME> TO ROLE <GENAI_ROLE_NAME>；

重要事项

• 用户上下文必须对支撑语义视图的物理表具有 SELECT 权限。仅授予语义视图访问权限会导致查询执行时出现运行时错误。

• 如果安全策略不允许直接访问基表，请在物理表上放置 Snowflake 安全视图，并在安全视图之上构建语义视图。由于 Snowflake 视图以所有者权限执行，用户 GenAI 角色只需对安全视图和语义视图具有 SELECT 权限，而不需要对原始表具有权限。

• Snowflake OAuth 令牌会在授权的确切时刻捕获并绑定到用户的 DEFAULT_ROLE 和 DEFAULT_WAREHOUSE。如果用户在配置这些默认值之前授权连接，会话将被锁定到错误的角色（当时可用的角色），如果目标角色没有所需权限，查询执行将失败。在 Gemini Enterprise 中授权之前，请先设置默认值（如下所示）。授权连接：让用户在 Gemini Enterprise 中执行一次性的 OAuth 授权。（可选）恢复默认值：授权后，用户可以安全地将 Snowflake 的默认角色和仓库恢复为标准操作偏好。

ALTER USER <BUSINESS_USER_NAME> SET DEFAULT_ROLE = '<GENAI_ROLE_NAME>'；ALTER USER <BUSINESS_USER_NAME> SET DEFAULT_WAREHOUSE = '<WAREHOUSE_NAME>'；

第 3 步：在 Snowflake 中部署 Cortex 代理和 MCP 服务器

接下来，使用语义视图在 Snowflake 中创建受管控的 Cortex 代理，然后通过 CORTEX_AGENT_RUN 工具类型部署一个 MCP 服务器以暴露代理。这种方法将 SQL 生成和查询执行委托给 Snowflake，同时强制实施调用者的 OAuth 身份和角色权限。

使用数据库 <DATABASE_NAME>；使用模式 <SCHEMA_NAME>；使用角色 SYSADMIN；-- 1. 在语义视图上创建 Cortex Agent 创建或替换代理 <AGENT_NAME> 注释 = '用于业务查询的受管数据助手' 配置文件 = '{"cortex_analyst": {"semantic_models": ["<DATABASE_NAME>.<SCHEMA_NAME>.<SEMANTIC_VIEW_NAME>"]}}'；-- 2. 部署暴露 Cortex Agent 的 MCP 服务器 创建或替换 MCP 服务器 <SERVER_NAME> 来自规范 $$ 工具：- 名称："run_cortex_agent" 类型："CORTEX_AGENT_RUN" 标识符："<DATABASE_NAME>.<SCHEMA_NAME>.<AGENT_NAME>" 标题："Cortex Agent 助手" 描述："将自然语言问题转换为 SQL，并使用调用者的 OAuth 会话权限直接在 Snowflake 中执行。" $$；

隔离与治理注意事项：通过使用 CORTEX_AGENT_RUN，查询生成和执行将完全保留在 Snowflake 的受管安全边界内。避免向外部客户端暴露 SYSTEM_EXECUTE_SQL 等原始执行工具，这样做会引入不必要的安全风险并绕过集中式 Cortex Agent 治理。

步骤 4：定位和构建 MCP 服务器 URL

ADK 代码需要 Snowflake 管理的 MCP 服务器的端点 URL。现在服务器已部署，您需要先授予服务器访问权限，然后可以使用 Snowflake 工作表中的以下查询获取其精确端点 URL：

-- 1. 授予 Cortex Agent 和 MCP 服务器的访问权限 使用角色 SECURITYADMIN；授予角色 <GENAI_ROLE_NAME> 使用代理 <DATABASE_NAME>.<SCHEMA_NAME>.<AGENT_NAME> 的权限；授予角色 <GENAI_ROLE_NAME> 使用 MCP 服务器 <DATABASE_NAME>.<SCHEMA_NAME>.<SERVER_NAME> 的权限；-- 2. 查看 MCP URL 描述 MCP 服务器 <DATABASE_NAME>.<SCHEMA_NAME>.<SERVER_NAME>；-- 或：DESC MCP 服务器 <DATABASE_NAME>.<SCHEMA_NAME>.<SERVER_NAME>；

在命令输出中，检查元数据属性以定位端点链接。URL 结构如下：

https://<org_name>-<account_name>.snowflakecomputing.com/api/v2/databases/<database_name>/schemas/<schema_name>/mcp-servers/<server_name>

步骤 5：通过 Google ADK 连接

创建 MCP 服务器并解析其 URL 后，使用 Google Agent 开发工具包 (ADK) 编写代理代码。在 Python 代码中，定义代理用于提取用户凭证的唯一标识符 AUTH_ID（例如 snowflake_oauth_conn）。

import os import re from typing import Optional from google.adk.agents.llm_agent import LlmAgent from google.adk.tools.mcp_tool.mcp_toolset import McpToolset from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams from google.adk.tools.tool_context import ToolContext # 在代码中首先定义授权ID # 注册代理时将使用这个确切名称 # Gemini Enterprise 会将授权令牌映射到 ToolContext 中的此标识符 AUTH_ID = "snowflake_oauth_conn" # 1. OAuth 头信息提取 # 定义回调函数以动态提取 Gemini Enterprise 注入的用户 OAuth 令牌 # 并生成与 Snowflake MCP 认证所需的授权头信息 def get_auth_headers(tool_context: Optional[ToolContext] = None) -> dict: """ OAuth 处理回调：动态提取 Gemini Enterprise 注入的用户 OAuth 令牌，并生成与 Snowflake MCP 认证所需的头信息 """ token = None if tool_context and hasattr(tool_context, "state"): try: # 解析由 Gemini Enterprise 注入的包含授权状态的 state 字典 state_dict = tool_context.state.to_dict() if hasattr(tool_context.state, 'to_dict') else tool_context.state pattern = re.compile(fr".*{re.escape(AUTH_ID)}.*", re.IGNORECASE) # 查找与我们的 Auth ID（snowflake_oauth_conn）匹配的活动令牌 matching_keys = [k for k in state_dict.keys() if pattern.match(k)] if matching_keys: token = state_dict.get(matching_keys[0]) except Exception as e: print(f"警告：读取 ToolContext OAuth 状态失败：{e}") # 返回授权头信息。如果 token 为 None，MCP 请求将失败 return { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json, text/event-stream", } # 2. MCP 初始化 # 直接在代码中定义从 'DESCRIBE MCP SERVER' 获取的 Snowflake MCP URL： MCP_SERVER_URL = "<YOUR_MCP_SERVER_URL>" # 替换为从 'DESCRIBE MCP SERVER' 获取的实际 URL # 使用此 URL 初始化 Snowflake MCP 工具集 snowflake_mcp = McpToolset( connection_params=StreamableHTTPConnectionParams( url=MCP_SERVER_URL ), header_provider=get_auth_headers ) # 3. 构建代理 # 使用 Gemini 2.5 Flash 模型构建 LlmAgent 并附加 Snowflake MCP 工具集 root_agent = LlmAgent( model="gemini-2.5-flash", name="enterprise_data_analyst", instruction=( "你是一名企业数据分析师。使用 Snowflake MCP 服务器上的 enterprise_data_agent 工具 " "回答业务问题。将用户的自然语言问题传递给工具，工具会将其转换为 SQL 并 " "在 Snowflake 的受控边界内直接执行，然后清晰地向用户呈现结果。" ), tools=[snowflake_mcp] )

注意：上面的 Python 代码片段表示代理设置的纯/基础版本。根据组织的具体需求，应在该模板基础上增加健壮性（例如自定义错误处理、结构化模式、自定义提示、配置和额外工具包装器）。

第6步：项目结构和部署

要将 ADK 代理部署到 Agent Platform 的 Agent Runtime，请按以下目录结构组织文件：

/think

snowflake_agent/ ├── agent.py # 代理定义和 MCP 连接代码（来自步骤 5） ├── __init__.py # 包初始化文件，用于暴露代理对象 ├── .env # 环境配置（Google Cloud 项目等） └── requirements.txt # 项目依赖列表（google-adk 等）

• 文件详情

• __init__.py: 向部署 CLI 暴露 root_agent 对象：

Snowflake ADK 代理包 from .agent import root_agent

• .env: 声明环境配置：

GOOGLE_CLOUD_PROJECT="<YOUR_GCP_PROJECT_ID>" GOOGLE_CLOUD_LOCATION="<LOCATION>" GOOGLE_GENAI_USE_VERTEXAI="TRUE"

• requirements.txt: 指定依赖项：

google-adk[mcp]==2.3.0 google-genai==2.10.0 pydantic-settings==2.14.2 google-cloud-aiplatform[agent_engines]>=1.50.0

1. 通过 CLI 部署

无需运行复杂的自定义 Python 脚本或手动构建容器，您可以使用官方 Google Agents CLI (agents-cli) 进行部署。CLI 会打包您的代码、设置服务器并处理部署流程：

将代理部署到 Google Cloud Agent Runtime agents-cli deploy --deployment-target agent_runtime

CLI 会自动处理部署生命周期。完成后，它会输出唯一的项目资源路径（例如：projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID），您将在下一步使用该路径注册代理。

提示：如需了解 ADK 代理部署的更多信息和高级配置选项，请参考：

• ADK 代理运行时部署指南
• ADK 部署参考文档

第 7 步：在 Gemini Enterprise 中注册自定义代理

当代理代码准备就绪并部署完成后，在 Google Cloud Console 中注册代理。在此过程中，您将链接代理配置并映射从 Snowflake 安全集成中获取的 OAuth 参数。

• 从 Snowflake 提取 OAuth 端点

在 Snowflake 中运行以下查询以提取精确的 Token 和 Authorization 端点：

DESCRIBE SECURITY INTEGRATION <SECURITY_INTEGRATION_NAME>;

在结果中找到以下属性的值：

• OAUTH_AUTHORIZATION_ENDPOINT : 您的 Snowflake 授权 URI。
• OAUTH_TOKEN_ENDPOINT : 您的 Snowflake Token URI。

1. Google Cloud Console 中的注册步骤

• 转至控制台中的 Gemini Enterprise 页面。
• 选择要注册代理的应用程序。
• 在左侧导航栏中点击 Agents，然后点击 + 添加代理。
• 选择 通过 Agent Platform 添加自定义代理。此时会显示授权设置。
• 设置 Snowflake OAuth 连接：点击 添加授权。授权名称：输入在 ADK 代码中定义的精确 ID（例如：snowflake_oauth_conn）。这确保凭证能正确映射到代码中的 AUTH_ID。客户端 ID & 客户端密钥：输入通过 SYSTEM$SHOW_OAUTH_CLIENT_SECRETS 查询从 Snowflake 获取的凭证。Token URI：输入从 DESCRIBE SECURITY INTEGRATION 提取的精确 Token 端点值。授权 URI：从 DESCRIBE SECURITY INTEGRATION 提取授权端点值，并在末尾添加 ?response_type=code（同时包含您的客户端 ID 和范围查询参数，例如）：

https://<org_name>-<account_name>.snowflakecomputing.com/oauth/authorize?client_id=YOUR_CLIENT_ID&redirect_uri=https%3A%2F%2Fvertexaisearch.cloud.google.com%2Foauth-redirect&scope=session%3Arole-any&response_type=code

IMPORTANT 确保授权 URI 以 &response_type=code 结尾（如果不存在其他参数，则使用 ?response_type=code）。该参数由 Google Cloud 要求，用于触发 OAuth 2.0 授权码流程。

• 点击 完成，然后点击 下一步。

1. 配置代理属性：

1. 点击 创建 以完成注册。

标识符对齐说明

• ADK 代码 AUTH_ID <==> Google Cloud 授权名称：Python 脚本中的 AUTH_ID 必须与控制台注册时输入的授权名称/ID 完全一致。由于我们先编写代码，因此在注册时会复用代码中定义的标识符。

• Snowflake 安全集成名称（例如：<SECURITY_INTEGRATION_NAME>）：这是 Snowflake 侧的数据库对象名称，无需与 Google Cloud 授权名称或 ADK Python 代码中的 AUTH_ID 匹配。

入门指南

通过结合 Gemini 模型的推理能力、ADK 的编排功能以及 Snowflake 的受控数据层，您可以安全地将 AI 引入数据。此架构提供了一个智能代理助手，在遵守您的安全边界的同时交付商业价值。

发布于

• 数据分析