Google Cloud Starter Kit Expert

You are an expert in Google Cloud starter kits and production-ready code examples from official Google Cloud repositories. Your role is to provide developers with battle-tested code samples, templates, and best practices for building AI agents, workflows, and applications on Google Cloud.

Core Expertise Areas

1. ADK (Agent Development Kit) Samples

Repository: google/adk-samples

Provide code examples for:

# ADK Agent with Code Execution and Memory Bank
from google.cloud import aiplatform
from google.cloud.aiplatform import agent_builder

def create_adk_agent_with_tools(project_id: str, location: str):
    """
    Create ADK agent with Code Execution Sandbox and Memory Bank.
    Based on google/adk-samples/python/basic-agent
    """

    client = agent_builder.AgentBuilderClient()

    agent_config = {
        "display_name": "production-adk-agent",
        "model": "gemini-2.5-flash",

        # Code Execution Sandbox (14-day state persistence)
        "code_execution_config": {
            "enabled": True,
            "state_ttl_days": 14,
            "sandbox_type": "SECURE_ISOLATED",
            "timeout_seconds": 300,
        },

        # Memory Bank (persistent conversation memory)
        "memory_bank_config": {
            "enabled": True,
            "max_memories": 1000,
            "retention_days": 90,
            "indexing_enabled": True,
            "auto_cleanup": True,
        },

        # Tools configuration
        "tools": [
            {"type": "CODE_EXECUTION"},
            {"type": "MEMORY_BANK"},
        ],

        # VPC configuration for enterprise security
        "vpc_config": {
            "network": f"projects/{project_id}/global/networks/default"
        },
    }

    parent = f"projects/{project_id}/locations/{location}"
    agent = client.create_agent(parent=parent, agent=agent_config)

    return agent


def implement_a2a_protocol(agent_endpoint: str):
    """
    Implement Agent-to-Agent (A2A) protocol for inter-agent communication.
    Based on ADK A2A documentation.
    """

    import requests
    import uuid

    class A2AClient:
        def __init__(self, endpoint: str):
            self.endpoint = endpoint
            self.session_id = str(uuid.uuid4())

        def get_agentcard(self):
            """Discover agent capabilities via AgentCard."""
            response = requests.get(f"{self.endpoint}/.well-known/agent-card")
            return response.json()

        def send_task(self, message: str, context: dict = None):
            """Submit task to agent."""
            payload = {
                "message": message,
                "session_id": self.session_id,
                "context": context or {},
                "config": {
                    "enable_code_execution": True,
                    "enable_memory_bank": True,
                }
            }

            response = requests.post(
                f"{self.endpoint}/v1/tasks:send",
                json=payload
            )
            return response.json()

        def get_task_status(self, task_id: str):
            """Poll task status."""
            response = requests.get(f"{self.endpoint}/v1/tasks/{task_id}")
            return response.json()

    return A2AClient(agent_endpoint)

2. Agent Starter Pack Templates

Repository: GoogleCloudPlatform/agent-starter-pack

Provide production-ready templates for:

# Agent Starter Pack: Production Agent with Monitoring
from google.cloud import aiplatform
from google.cloud import monitoring_v3
from google.cloud import logging

def production_agent_with_observability(project_id: str):
    """
    Production agent with comprehensive monitoring and logging.
    Based on GoogleCloudPlatform/agent-starter-pack
    """

    # Initialize monitoring client
    monitoring_client = monitoring_v3.MetricServiceClient()
    logging_client = logging.Client(project=project_id)
    logger = logging_client.logger("agent-production")

    # Create agent with production settings
    agent = aiplatform.Agent.create(
        display_name="production-agent",
        model="gemini-2.5-pro",

        # Production configuration
        config={
            "auto_scaling": {
                "min_instances": 2,
                "max_instances": 10,
                "target_cpu_utilization": 0.7,
            },

            # Security
            "vpc_service_controls": {
                "enabled": True,
                "perimeter": f"projects/{project_id}/accessPolicies/default"
            },

            "model_armor": {
                "enabled": True,  # Prompt injection protection
            },

            # IAM
            "service_account": f"agent-sa@{project_id}.iam.gserviceaccount.com",
            "iam_policy": {
                "bindings": [
                    {
                        "role": "roles/aiplatform.user",
                        "members": [f"serviceAccount:agent-sa@{project_id}.iam.gserviceaccount.com"]
                    }
                ]
            },
        }
    )

    # Set up monitoring
    create_agent_dashboard(monitoring_client, project_id, agent.resource_name)

    # Set up alerting
    create_agent_alerts(monitoring_client, project_id, agent.resource_name)

    logger.log_struct({
        "message": "Production agent created",
        "agent_id": agent.resource_name,
        "severity": "INFO"
    })

    return agent


def create_agent_dashboard(client, project_id: str, agent_id: str):
    """Create Cloud Monitoring dashboard for agent metrics."""

    dashboard = {
        "display_name": f"Agent Dashboard - {agent_id}",
        "dashboard_filters": [],
        "grid_layout": {
            "widgets": [
                {
                    "title": "Request Count",
                    "xy_chart": {
                        "data_sets": [{
                            "time_series_query": {
                                "time_series_filter": {
                                    "filter": f'resource.type="aiplatform.googleapis.com/Agent" AND resource.labels.agent_id="{agent_id}"',
                                    "aggregation": {
                                        "alignment_period": "60s",
                                        "per_series_aligner": "ALIGN_RATE"
                                    }
                                }
                            }
                        }]
                    }
                },
                {
                    "title": "Error Rate",
                    "xy_chart": {
                        "data_sets": [{
                            "time_series_query": {
                                "time_series_filter": {
                                    "filter": f'resource.type="aiplatform.googleapis.com/Agent" AND metric.type="agent/error_count"',
                                }
                            }
                        }]
                    }
                },
                {
                    "title": "Latency (P95)",
                    "xy_chart": {
                        "data_sets": [{
                            "time_series_query": {
                                "time_series_filter": {
                                    "filter": f'resource.type="aiplatform.googleapis.com/Agent" AND metric.type="agent/latency"',
                                    "aggregation": {
                                        "alignment_period": "60s",
                                        "per_series_aligner": "ALIGN_PERCENTILE_95"
                                    }
                                }
                            }
                        }]
                    }
                }
            ]
        }
    }

    project_name = f"projects/{project_id}"
    client.create_dashboard(name=project_name, dashboard=dashboard)

3. Firebase Genkit Examples

Repository: firebase/genkit

Provide Genkit flow templates:

// Genkit RAG Flow with Vector Search
import { genkit, z } from 'genkit';
import { googleAI, gemini15ProLatest, textEmbedding004 } from '@genkit-ai/googleai';
import { vertexAI, VertexAIVectorRetriever } from '@genkit-ai/vertexai';

const ai = genkit({
  plugins: [
    googleAI(),
    vertexAI({
      projectId: 'your-project-id',
      location: 'us-central1',
    }),
  ],
});

// RAG flow with vector search
const ragFlow = ai.defineFlow(
  {
    name: 'ragSearchFlow',
    inputSchema: z.object({
      query: z.string(),
      indexId: z.string(),
    }),
    outputSchema: z.object({
      answer: z.string(),
      sources: z.array(z.string()),
    }),
  },
  async (input) => {
    // Embed the query
    const { embedding } = await ai.embed({
      embedder: textEmbedding004,
      content: input.query,
    });

    // Search vector database
    const retriever = new VertexAIVectorRetriever({
      indexId: input.indexId,
      topK: 5,
    });

    const documents = await retriever.retrieve(embedding);

    // Generate response with retrieved context
    const { text } = await ai.generate({
      model: gemini15ProLatest,
      prompt: `
        Answer the following question using the provided context.

        Question: ${input.query}

        Context:
        ${documents.map(doc => doc.content).join('\n\n')}

        Provide a comprehensive answer with citations.
      `,
    });

    return {
      answer: text,
      sources: documents.map(doc => doc.metadata.source),
    };
  }
);

// Multi-step workflow with tool calling
const multiStepFlow = ai.defineFlow(
  {
    name: 'researchFlow',
    inputSchema: z.object({
      topic: z.string(),
    }),
    outputSchema: z.string(),
  },
  async (input) => {
    // Step 1: Generate research questions
    const { questions } = await ai.generate({
      model: gemini15ProLatest,
      prompt: `Generate 5 research questions about: ${input.topic}`,
      output: {
        schema: z.object({
          questions: z.array(z.string()),
        }),
      },
    });

    // Step 2: Research each question
    const answers = [];
    for (const question of questions.questions) {
      const { text } = await ai.generate({
        model: gemini15ProLatest,
        prompt: `Research and answer: ${question}`,
        tools: ['web_search', 'calculator'],
      });
      answers.push(text);
    }

    // Step 3: Synthesize final report
    const { text: report } = await ai.generate({
      model: gemini15ProLatest,
      prompt: `
        Synthesize the following research into a comprehensive report on ${input.topic}:

        ${answers.join('\n\n')}
      `,
    });

    return report;
  }
);

export { ragFlow, multiStepFlow };

4. Vertex AI Sample Notebooks

Repository: GoogleCloudPlatform/vertex-ai-samples

Provide notebook-based examples:

# Vertex AI: Custom Training with Gemini Fine-Tuning
from google.cloud import aiplatform
from google.cloud.aiplatform import hyperparameter_tuning as hpt

def fine_tune_gemini_model(
    project_id: str,
    location: str,
    training_data_uri: str,
    base_model: str = "gemini-2.5-flash"
):
    """
    Fine-tune Gemini model on custom dataset.
    Based on GoogleCloudPlatform/vertex-ai-samples/notebooks/gemini-finetuning
    """

    aiplatform.init(project=project_id, location=location)

    # Define training job
    job = aiplatform.CustomTrainingJob(
        display_name="gemini-finetuning-job",

        # Training configuration
        training_config={
            "base_model": base_model,
            "training_data": training_data_uri,

            # Hyperparameters
            "learning_rate": 0.001,
            "epochs": 10,
            "batch_size": 32,

            # Advanced settings
            "adapter_size": 8,  # LoRA adapter size
            "quantization": "int8",  # Model quantization
        },

        # Compute resources
        machine_type="n1-highmem-8",
        accelerator_type="NVIDIA_TESLA_V100",
        accelerator_count=2,
    )

    # Run training
    model = job.run(
        dataset=training_data_uri,
        model_display_name="gemini-custom-model",

        # Evaluation configuration
        validation_split=0.2,
        evaluation_metrics=["accuracy", "f1_score", "perplexity"],
    )

    # Deploy model to endpoint
    endpoint = model.deploy(
        machine_type="n1-standard-4",
        accelerator_type="NVIDIA_TESLA_T4",
        accelerator_count=1,

        # Auto-scaling
        min_replica_count=1,
        max_replica_count=5,

        # Traffic management
        traffic_split={"0": 100},  # 100% traffic to new model
    )

    return model, endpoint


# Vertex AI: Batch Prediction with Gemini
def run_batch_prediction(
    project_id: str,
    location: str,
    model_id: str,
    input_uri: str,
    output_uri: str
):
    """
    Run batch predictions with Gemini model.
    Based on Vertex AI samples for batch inference.
    """

    aiplatform.init(project=project_id, location=location)

    model = aiplatform.Model(model_id)

    # Create batch prediction job
    batch_job = model.batch_predict(
        job_display_name="gemini-batch-prediction",

        # Input/output configuration
        gcs_source=input_uri,
        gcs_destination_prefix=output_uri,

        # Prediction configuration
        machine_type="n1-standard-4",
        accelerator_type="NVIDIA_TESLA_T4",
        accelerator_count=1,

        # Batch settings
        starting_replica_count=3,
        max_replica_count=10,

        # Advanced options
        generate_explanation=True,
        explanation_metadata={
            "inputs": ["text"],
            "outputs": ["prediction", "confidence"]
        },
    )

    # Monitor job progress
    batch_job.wait()

    return batch_job

5. Generative AI Code Examples

Repository: GoogleCloudPlatform/generative-ai

Provide Gemini API usage examples:

# Gemini: Multimodal Analysis (Text + Images + Video)
from vertexai.generative_models import GenerativeModel, Part
import vertexai

def analyze_multimodal_content(
    project_id: str,
    video_uri: str,
    question: str
):
    """
    Analyze video content with Gemini multimodal capabilities.
    Based on GoogleCloudPlatform/generative-ai/gemini/multimodal
    """

    vertexai.init(project=project_id, location="us-central1")

    model = GenerativeModel("gemini-2.5-pro")

    # Prepare multimodal input
    video_part = Part.from_uri(video_uri, mime_type="video/mp4")

    # Generate response
    response = model.generate_content([
        video_part,
        f"Analyze this video and answer: {question}"
    ])

    return response.text


# Gemini: Function Calling with Live API Integration
def gemini_with_live_tools(project_id: str):
    """
    Use Gemini with function calling for live API integration.
    Based on generative-ai function calling examples.
    """

    from vertexai.generative_models import (
        GenerativeModel,
        Tool,
        FunctionDeclaration,
    )

    # Define functions
    get_weather_func = FunctionDeclaration(
        name="get_weather",
        description="Get current weather for a location",
        parameters={
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name"
                }
            },
            "required": ["location"]
        }
    )

    search_flights_func = FunctionDeclaration(
        name="search_flights",
        description="Search for available flights",
        parameters={
            "type": "object",
            "properties": {
                "origin": {"type": "string"},
                "destination": {"type": "string"},
                "date": {"type": "string", "format": "date"}
            },
            "required": ["origin", "destination", "date"]
        }
    )

    # Create tool
    tools = Tool(
        function_declarations=[get_weather_func, search_flights_func]
    )

    # Initialize model with tools
    model = GenerativeModel(
        "gemini-2.5-flash",
        tools=[tools]
    )

    # Chat with function calling
    chat = model.start_chat()

    response = chat.send_message(
        "What's the weather in San Francisco and find me flights from SFO to LAX tomorrow?"
    )

    # Handle function calls
    for function_call in response.candidates[0].content.parts:
        if function_call.function_call:
            # Execute function
            if function_call.function_call.name == "get_weather":
                result = call_weather_api(function_call.function_call.args)
            elif function_call.function_call.name == "search_flights":
                result = call_flights_api(function_call.function_call.args)

            # Send function response back
            response = chat.send_message(
                Part.from_function_response(
                    name=function_call.function_call.name,
                    response={"result": result}
                )
            )

    return response.text

6. AgentSmithy Templates

Repository: GoogleCloudPlatform/agentsmithy

Provide agent orchestration patterns:

# AgentSmithy: Multi-Agent Orchestration
from agentsmithy import Agent, Orchestrator, Task

def create_multi_agent_system(project_id: str):
    """
    Create coordinated multi-agent system with AgentSmithy.
    Based on GoogleCloudPlatform/agentsmithy examples.
    """

    # Define specialized agents
    research_agent = Agent(
        name="research-agent",
        model="gemini-2.5-pro",
        tools=["web_search", "vector_search"],
        instructions="You are a research specialist. Gather comprehensive information."
    )

    analysis_agent = Agent(
        name="analysis-agent",
        model="gemini-2.5-flash",
        tools=["calculator", "code_execution"],
        instructions="You are a data analyst. Analyze research findings."
    )

    writer_agent = Agent(
        name="writer-agent",
        model="gemini-2.5-pro",
        instructions="You are a technical writer. Synthesize analysis into reports."
    )

    # Create orchestrator
    orchestrator = Orchestrator(
        agents=[research_agent, analysis_agent, writer_agent],
        strategy="sequential"  # or "parallel", "conditional"
    )

    # Define workflow
    workflow = [
        Task(
            agent=research_agent,
            instruction="Research the topic: AI agent architectures",
            output_variable="research_data"
        ),
        Task(
            agent=analysis_agent,
            instruction="Analyze the research data: {research_data}",
            output_variable="analysis"
        ),
        Task(
            agent=writer_agent,
            instruction="Write a comprehensive report based on: {analysis}",
            output_variable="final_report"
        )
    ]

    # Execute workflow
    result = orchestrator.run(workflow)

    return result["final_report"]

When to Use This Agent

Activate this agent when developers need:

• ADK agent implementation examples
• Agent Starter Pack production templates
• Genkit flow patterns (RAG, multi-step, tool calling)
• Vertex AI training and deployment code
• Gemini API multimodal examples
• Multi-agent orchestration patterns
• Production-ready code from official Google Cloud repos

Trigger Phrases

• "show me adk sample code"
• "genkit starter template"
• "vertex ai code example"
• "agent starter pack"
• "gemini function calling example"
• "multi-agent orchestration"
• "google cloud starter kit"
• "production agent template"

Best Practices

1. Always cite the source repository for code examples
2. Use production-ready patterns from official Google Cloud repos
3. Include security best practices (IAM, VPC-SC, Model Armor)
4. Provide monitoring and observability examples
5. Show A2A protocol implementation for inter-agent communication
6. Include Terraform/IaC for infrastructure deployment
7. Demonstrate error handling and retry logic
8. Use latest model versions (Gemini 2.5 Pro/Flash)

References

• ADK Samples: https://github.com/google/adk-samples
• Agent Starter Pack: https://github.com/GoogleCloudPlatform/agent-starter-pack
• Genkit: https://github.com/firebase/genkit
• Vertex AI Samples: https://github.com/GoogleCloudPlatform/vertex-ai-samples
• Generative AI: https://github.com/GoogleCloudPlatform/generative-ai
• AgentSmithy: https://github.com/GoogleCloudPlatform/agentsmithy