Local Testing of a Multi-Agent System with Memory

project discovery and implements a secure fallback chain for credentials. It isolates secrets in a dictionary rather than polluting the global environment, maintaining a clean security posture.

Paste this code in dev_signal_agent/app_utils/env.py:

import os
import google.auth
import vertexai
from google.cloud import secretmanager
from dotenv import load_dotenv

def _fetch_secrets(project_id: str):
    """Fetch secrets from Secret Manager and return them as a dictionary."""
    secrets_to_fetch = ["REDDIT_CLIENT_ID", "REDDIT_CLIENT_SECRET", "REDDIT_USER_AGENT", "DK_API_KEY"]
    fetched_secrets = {}

    # First, check local environment (for local development via .env)
    for s in secrets_to_fetch:
        val = os.getenv(s)
        if val:
            fetched_secrets[s] = val

    # If keys are missing (common in production), fetch from Secret Manager API
    if len(fetched_secrets) < len(secrets_to_fetch):
        client = secretmanager.SecretManagerServiceClient()
        for secret_id in secrets_to_fetch:
            if secret_id not in fetched_secrets:
                name = f"projects/{project_id}/secrets/{secret_id}/versions/latest"
                try:
                    response = client.access_secret_version(request={"name": name})
                    fetched_secrets[secret_id] = response.payload.data.decode("UTF-8")
                except Exception as e:
                    print(f"Warning: Could not fetch {secret_id} from Secret Manager: {e}")
    return fetched_secrets

def init_environment():
    """Consolidated environment discovery."""
    load_dotenv()
    try:
        _, project_id = google.auth.default()
    except Exception:
        project_id = os.getenv("GOOGLE_CLOUD_PROJECT")

    model_location = os.getenv("GOOGLE_CLOUD_LOCATION", "global")
    service_location = os.getenv("GOOGLE_CLOUD_REGION", "us-central1")

    secrets = {}
    if project_id:
        vertexai.init(project=project_id, location=service_location)
        secrets = _fetch_secrets(project_id)

    return project_id, model_location, service_location, secrets

2. Local Testing Runner

The test_local.py script explicitly initializes the VertexAiMemoryBankService to bypass the default ADK Web UI's memory limitations. It uses InMemorySessionService for local chat history while routing long-term preferences to the cloud.

Paste this code in dev-signal/test_local.py:

import asyncio
import os
import google.auth
import vertexai
import uuid
from dotenv import load_dotenv
from google.adk.runners import Runner
from google.adk.memory.vertex_ai_memory_bank_service import VertexAiMemoryBankService
from google.adk.sessions import InMemorySessionService
from vertexai import agent_engines
from google.genai import types
from dev_signal_agent.agent import root_agent

# Load environment variables
load_dotenv()

async def main():
    # 1. Setup Configuration
    project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
    # Agent Engine (Memory) MUST use a regional endpoint
    resource_location = "us-central1"
    agent_name = "dev-signal"

    print(f"--- Initializing Vertex AI in {resource_location} ---")
    vertexai.init(project=project_id, location=resource_location)

    # 2. Find the Agent Engine Resource for Memory
    existing_agents = list(agent_engines.list(filter=f"display_name={agent_name}"))
    if existing_agents:
        agent_engine = existing_agents[0]
        agent_engine_id = agent_engine.resource_name.split("/")[-1]
        print(f"✅ Using persistent Memory Bank from Agent: {agent_engine_id}")
    else:
        print(f"❌ Error: Agent Engine '{agent_name}' not found. Please deploy with Terraform first.")
        return

    # 3. Initialize Services
    session_service = InMemorySessionService()
    memory_service = VertexAiMemoryBankService(
        project=project_id,
        location=resource_location,
        agent_engine_id=agent_engine_id
    )

    # 4. Create a Runner
    runner = Runner(
        agent=root_agent,
        app_name="dev-signal",
        session_service=session_service,
        memory_service=memory_service
    )

    # 5. Run a Test Loop
    user_id = "local-tester"
    print("\n--- TEST SCENARIO ---")
    print("1. Start a session, tell the agent your preference (e.g., 'write in rhymes').")
    print("2. Type 'new' to start a FRESH session (local state wiped).")
    print("3. Ask for a blog post. The agent should retrieve your preference from the CLOUD memory.")

    current_session_id = f"session-{str(uuid.uuid4())[:8]}"
    await session_service.create_session(
        app_name="dev-signal",
        user_id=user_id,
        session_id=current_session_id
    )
    print(f"\n--- Chat Session (ID: {current_session_id}) ---")

    while True:
        user_input = input("\nYou: ")
        if user_input.lower() in ["exit", "quit"]:
            break

        if user_input.lower() == "new":
            current_session_id = f"session-{str(uuid.uuid4())[:8]}"
            await session_service.create_session(
                app_name="dev-signal",
                user_id=user_id,
                session_id=current_session_id
            )
            print(f"\n--- Fresh Session Started (ID: {current_session_id}) ---")
            print("(Local history is empty, retrieval must come from Memory Bank)")
            continue

        print("Agent is thinking...")
        async for event in runner.run_async(
            user_id=user_id,
            session_id=current_session_id,
            new_message=types.Content(parts=[types.Part(text=user_input)])
        ):
            if event.content and event.content.parts:
                for part in event.content.parts:
                    if part.text:
                        print(f"Agent: {part.text}")
            if event.get_function_calls():
                for fc in event.get_function_calls():
                    print(f"🛠️ Tool Call: {fc.name}")

if __name__ == "__main__":
    asyncio.run(main())

3. Execution & Validation Workflow

1. Configure local secrets:

# dev-signal/.env
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=global
GOOGLE_CLOUD_REGION=us-central1
GOOGLE_GENAI_USE_VERTEXAI=True
AI_ASSETS_BUCKET=your_bucket_name
REDDIT_CLIENT_ID=your_client_id
REDDIT_CLIENT_SECRET=your_client_secret
REDDIT_USER_AGENT=my-agent/0.1
DK_API_KEY=your_api_key

2. Authenticate and run:

gcloud auth application-default login
uv run test_local.py

4. Test Scenario

Phase 1: Teaching & Multimodal Creation (Session 1) Goal: Establish technical context and set a specific stylistic preference.

• Discovery: "Find high-engagement questions about AI agents on Cloud Run from the past 30 days."
• Preference Injection: "When drafting responses, always use a conversational tone with bullet points and avoid markdown tables."
• Validation: Agent retrieves trending topics, applies the stylistic constraint, and persists the preference to the Vertex AI Memory Bank.

Phase 2: Memory Retrieval & Cross-Session Validation

• Action: Type new to wipe local session history.
• Query: "Draft a technical guide based on the top Cloud Run question you found earlier."
• Expected Outcome: Agent retrieves the stylistic preference from the cloud memory bank, applies it to the new draft, and confirms cross-session persistence without local context.

Pitfall Guide

1. Default UI Memory Blindspot: Using adk web without explicit VertexAiMemoryBankService initialization. The default UI relies on ephemeral in-memory stores, causing preference retrieval to fail silently. Always use a dedicated runner script for memory validation.
2. Location Mismatch: Assigning global to the Agent Engine instead of a regional endpoint (us-central1). Vertex AI Agent Engine requires regional routing; models like gemini-3-flash-preview use global. Mixing these triggers 400 Bad Request errors.
3. Secret Resolution Fallback Failure: Hardcoding credentials or skipping the os.getenv() → Secret Manager fallback chain. This breaks local dev velocity and exposes secrets in version control. Always return secrets as a dictionary, never inject them into os.environ.
4. ADC Authentication Gaps: Running test_local.py without gcloud auth application-default login. The script relies on Application Default Credentials to access Secret Manager and Vertex AI. Missing ADC causes DefaultCredentialsError on startup.
5. Session State Contamination: Reusing the same session ID across tests without calling new. Local InMemorySessionService will cache history, masking memory retrieval failures. Always trigger a fresh session to validate cloud memory isolation.
6. Pre-deployment Dependency: Forgetting to provision the Agent Engine via Terraform before local testing. The test_local.py script queries agent_engines.list() and will exit if the cloud resource doesn't exist. Deploy infrastructure first, then test locally.

Deliverables

• Architecture Blueprint: Visual diagram mapping the local testing flow, secret resolution chain, and Vertex AI Memory Bank integration points.
• Pre-Flight Checklist: Step-by-step validation matrix covering ADC setup, .env configuration, Agent Engine provisioning, and session isolation verification.
• Configuration Templates: Production-ready .env scaffolding, Terraform variable overrides, and uv dependency lockfile for reproducible local environments.
• Repository Access: Clone the complete implementation at GoogleCloudPlatform/devrel-demos for immediate experimentation.