部署到Agent Engine的ADK代理无法访问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

