Service Account Impersonation to Call Vertex AI running in a Different GCP Project from the ADK Agent #2882
Description
Describe the bug
I need to write a sample ADK Agent that runs in GCP Project A in a GKE Cluster, that uses Workload Identity, but that agent needs to use a Gemini Model via the Vertex AI API via a service account defined in a different GCP Project (let's call it GCP Project B).
Ideally I would use a JSON credential file that defines the service account impersonation config to prevent a bunch of boilerplate python code, but if I don't have access to do that in the GKE environment, and I need to instead perform the service account impersonation in python code, how can I pass that short-lived credential object into my ADK Agent?
Service Account Impersonation Sample Code
def impersonate_service_account(settings: Settings):
source_credentials, _ = default()
service_account_to_impersonate = settings.IMPERSONATE_SERVICE_ACCOUNT
impersonated_creds = impersonated_credentials.Credentials(
source_credentials=source_credentials,
target_principal=service_account_to_impersonate,
target_scopes=["https://www.googleapis.com/auth/cloud-platform"],
quota_project_id=settings.IMPERSONATE_GCP_PROJECT
)
impersonated_creds.refresh(Request())
return impersonated_credsI see that ADK instantiates a google genai Client object that is defined here, and accepts a credentials object: https://github.com/googleapis/python-genai/blob/main/google/genai/client.py#L128-L133
Class Client:
...
def __init__(
self,
*,
vertexai: Optional[bool] = None,
api_key: Optional[str] = None,
credentials: Optional[google.auth.credentials.Credentials] = None,
project: Optional[str] = None,
location: Optional[str] = None,
debug_config: Optional[DebugConfig] = None,
http_options: Optional[Union[HttpOptions, HttpOptionsDict]] = None,
):
...But the ADK code that instantiates this doesn't pass in a credentials object:
https://github.com/google/adk-python/blob/main/src/google/adk/models/google_llm.py#L166-L177
class Gemini(BaseLlm):
...
def api_client(self) -> Client:
"""Provides the api client.
Returns:
The api client.
"""
return Client(
http_options=types.HttpOptions(
headers=self._tracking_headers,
retry_options=self.retry_options,
)
)Expected behavior
I want to be able to authenticate using GCP Service Account impersonation in my ADK agent with as little custom code as possible.
Workaround
I believe this monkey patch approach will do the trick, but I haven't done extensive testing with it. My setup is:
- A GCP user account that only has Service Account Token Creator on the SA in GCP Project B
- My user account has no other permissions in GCP Project B.
- The SA in GCP Project B has Vertex AI User and nothing else.
- Vertex AI is disabled in GCP Project A.
- I have tested this from a terminal where I'm logged in as the GCP user that mentioned in bullet one.
- I have not yet tested in a GKE cluster.
import vertexai
from vertexai.preview import reasoning_engines
from google.auth.credentials import Credentials
from google.genai import _api_client
from typing import Tuple, Union
from google.adk.agents import Agent
from google.adk.tools import google_search
TARGET_GCP_PROJECT_ID = "**********"
LOCATION = "us-central1"
IMPERSONATED_SERVICE_ACCOUNT = "***********[email protected]"
SCOPE = "https://www.googleapis.com/auth/cloud-platform"
def get_impersonated_credentials(
impersonated_service_account: str, scope: str
):
from google.auth import impersonated_credentials
import google.auth.transport.requests
"""
Use a service account (SA1) to impersonate another service account (SA2)
and obtain an ID token for the impersonated account.
To obtain a token for SA2, SA1 should have the
"roles/iam.serviceAccountTokenCreator" permission on SA2.
Args:
impersonated_service_account: The name of the privilege-bearing service account for whom the credential is created.
Examples: [email protected]
scope: Provide the scopes that you might need to request to access Google APIs,
depending on the level of access you need.
For this example, we use the cloud-wide scope and use IAM to narrow the permissions.
https://cloud.google.com/docs/authentication#authorization_for_services
For more information, see: https://developers.google.com/identity/protocols/oauth2/scopes
"""
# Construct the GoogleCredentials object which obtains the default configuration from your
# working environment.
credentials, project_id = google.auth.default()
# Create the impersonated credential.
target_credentials = impersonated_credentials.Credentials(
source_credentials=credentials,
target_principal=impersonated_service_account,
# delegates: The chained list of delegates required to grant the final accessToken.
# For more information, see:
# https://cloud.google.com/iam/docs/create-short-lived-credentials-direct#sa-credentials-permissions
# Delegate is NOT USED here.
delegates=[],
target_scopes=[scope],
lifetime=300,
)
# Get the OAuth2 token.
# Once you've obtained the OAuth2 token, use it to make an authenticated call
# to the target audience.
request = google.auth.transport.requests.Request()
target_credentials.refresh(request)
# The token field is target_credentials.token.
return target_credentials
# Monkey Patch for _api_client.py _load_auth() function, which the BaseApiClient does not
# allow for sending impersonated / custom credentials to as of 2025-08-14
def _load_auth(*, project: Union[str, None]) -> Tuple[Credentials, str]:
credentials = get_impersonated_credentials(IMPERSONATED_SERVICE_ACCOUNT, SCOPE)
return credentials, TARGET_GCP_PROJECT_ID
_api_client._load_auth = _load_auth
# Initialize Vertex AI must be initialized in order to support the impersonated SA credentials
# Because it will determine which GCP Project ID is used in the URL when calling
# the Gemini generateContent API endpoint
#
# There does not seem to be another way of handling this given the current state of the
# ADK/GenAI python libraries
vertexai.init(
project=TARGET_GCP_PROJECT_ID,
location=LOCATION,
staging_bucket=TARGET_GCP_PROJECT_ID,
credentials=get_impersonated_credentials(IMPERSONATED_SERVICE_ACCOUNT, SCOPE)
)
root_agent = Agent(
name='interesting_facts_agent',
model='gemini-2.5-flash-lite',
description=('Agent to give interesting facts.'),
instruction=('You are a helpful agent who can provide interesting facts.'),
tools=[google_search],
)
app = reasoning_engines.AdkApp(
agent=root_agent,
enable_tracing=False,
)
session = app.create_session(user_id="u_123")
for event in app.stream_query(
user_id="u_123",
session_id=session.id,
message="Tell me an interesting fact",
):
print(event)