You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

部署到Agent Engine的ADK代理无法访问Firestore:403权限问题求助

ADK代理部署到Agent Engine后Firestore 403权限问题排查思路
  • 确认Agent Engine实际使用的服务账号
    本地运行用的是你本地的凭据,但Agent Engine部署时用的是实例绑定的服务账号,别搞错对象。去实例详情里查这个账号,核对它的权限是否真的包含Cloud Datastore Owner、Firebase Admin,同时检查有没有组织级的IAM deny规则限制生效,或者角色添加后有没有等够生效时间。

  • 检查Firestore安全规则(Firebase项目需注意)
    就算服务账号有IAM权限,Firebase的安全规则可能拦截请求。去控制台看规则,确认是否允许服务账号访问users/{uid}/profile/info路径,比如规则如果只允许认证用户访问,服务账号调用没带认证信息就会被拦。

  • 验证Firestore客户端初始化逻辑
    你代码里直接用firestore.Client(project=PROJECT_ID),在Agent Engine环境中,是否需要指定服务账号密钥?或者环境有没有自动注入正确凭据?可以加日志打印db._credentials.service_account_email,确认当前用的是不是你配置的服务账号。

  • 核对资源路径和项目ID
    确认代码里的PROJECT_ID和Agent Engine部署的项目完全一致,有没有拼写错误;同时检查Firestore文档路径users/{user_uid}/profile/info的层级、拼写是否正确。

  • 本地测试服务账号权限
    下载该服务账号的密钥文件,用gcloud auth activate-service-account --key-file=xxx.json切换账号后,本地运行回调代码。如果本地也报错,就是权限配置问题;如果本地正常,就是Agent Engine环境的凭据注入有问题。

  • 查看详细错误日志
    去Cloud Logging找Agent Engine实例的日志,里面会有403错误的具体细节:哪个账号、哪个资源、缺失什么权限,比笼统的报错信息有用得多,比如可能是需要datastore.entities.get权限而角色没覆盖到。


你的before agent回调代码

from typing import Optional

from google.adk.agents.callback_context import CallbackContext
from google.genai import types
from google.cloud import firestore

PROJECT_ID = "my_project_id"
db = firestore.Client(project=PROJECT_ID)

def before_agent_callback(callback_context: CallbackContext) -> Optional[types.Content]:
    """
    极简前置代理回调:
    - 从会话中读取user_id(callback_context.session或_invocation_context.session)
    - 如果状态中没有'user:uid',则将其存入状态
    - 如果状态中缺少'user:profile',则获取Firestore中users/{uid}/profile/info文档
    - 将原始Firestore字典(或空字典)存入状态'user:profile'
    """
    state = callback_context.state

    if "user:uid" not in state or not state.get("user:uid"):
        session = getattr(callback_context, "session", None)
        if session is None:
            inv_ctx = getattr(callback_context, "_invocation_context", None)
            session = getattr(inv_ctx, "session", None) if inv_ctx else None
        if session and hasattr(session, "user_id"):
            state["user:uid"] = session.user_id

    user_uid = state.get("user:uid")
    if not user_uid:
        return None

    if "user:profile" not in state or not state.get("user:profile"):
        doc_ref = db.document(f"users/{user_uid}/profile/info")
        doc = doc_ref.get()
        state["user:profile"] = doc.to_dict() if doc.exists else {}

    return None

内容的提问来源于stack exchange,提问作者Victor

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.12 06:14:52