This is an example of how we can deployment agent runtime agent with environment variables. The key settings here is the "env_vars".
import asyncio
from urllib import response
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from vertexai import types as vertex_types
from google.genai import types
import vertexai
import agentplatform
from agentplatform import types # check below: this is my assumption
from agentplatform.frameworks import AdkApp
# 1. Define the callback function
def inspect_llm_input(callback_context: CallbackContext, llm_request: LlmRequest):
"""Callback triggered directly before ADK sends payload to the LLM."""
print("=" * 60)
print("INSPECTING INPUT PASSED TO LLM BY ADK")
print("=" * 60)
print(llm_request)
print("=" * 60 + "\n")
# Return None to continue execution normally
return None
# -------------------------------------------------------------
# 1. DEFINE THE TOOL FUNCTION
# ADK inspects docstrings and type hints to auto-generate
# the FunctionDeclaration for Gemini.
# -------------------------------------------------------------
def get_user_account_balance(user_id: str, include_pending: bool = False) -> dict:
"""Retrieves the current bank balance for a specific user ID.
Args:
user_id: The unique account identifier for the customer.
include_pending: Whether to include pending uncleared transactions.
"""
return {
"user_id": user_id,
"balance": "$4,250.00",
"pending": "$120.00" if include_pending else "$0.00"
}
# -------------------------------------------------------------
# 2. CREATE THE ADK AGENT
# -------------------------------------------------------------
banking_agent = Agent(
name="account_assistant",
model="gemini-2.5-flash",
instruction="You are a helpful customer support agent. Answer questions concisely.",
tools=[get_user_account_balance],
before_model_callback=inspect_llm_input
)
adk_app = AdkApp(agent=banking_agent)
# -------------------------------------------------------------
# 3. RUNNER & SESSION ORCHESTRATION
# -------------------------------------------------------------
async def main():
client = (
project="project-id",
location="us-central1",
)
# Define resource ID of your gateway
gateway_resource_id = (
"projects/project-id/locations/us-central1/agentGateways/my-first-gateway"
)
# Deploy agent with gateway configurations attached
remote_agent = client.runtimes.create(
agent=adk_app,
config={
"display_name": "Envs Serverless Agent",
"requirements": ["google-cloud-aiplatform[agent_engines,adk]"],
"staging_bucket": "gs://my-vertex-bucket-ai",
"identity_type": vertex_types.IdentityType.AGENT_IDENTITY,
"env_vars": {
"LOG_LEVEL": "INFO",
"API_BASE_URL": "https://example.com",
# Secret Manager reference, exposed as an env var.
# Do NOT turn this on, unless you have a secret in Secret Manager with this name and version.
#"DB_CREDENTIALS": {"secret": "my-secret-id", "version": "3"},
}
# Attach the Agent Gateway configuration block
# "agent_gateway_config": {
# "client_to_agent_config": {
# "agent_gateway": gateway_resource_id
# }
# },
},
)
print(remote_agent.api_resource.spec.deployment_spec.agent_gateway_config)
asyncio.run(main())
Please note we also have example of pulling secret from google secret manager. However, you must have these secret there otherwise you will get an error:
Comments