sequenceDiagram
autonumber
actor Developer as Attendee / Student
participant API as FastAPI Server
participant Gemini as Gemini 1.5 Flash
participant Tools as Domain Tools (Python)
participant Cloud as Google Cloud Run
Developer->>API: POST /api/navigate ("Flutter + Tailwind for civic tech")
API->>Gemini: Prompt + Declared Tool Definitions
Gemini->>Gemini: Decides tools needed
Gemini->>Tools: Function Call 1: search_tech_stack()
Tools-->>Gemini: Returns Firebase + Cloud Run + PostgreSQL
Gemini->>Tools: Function Call 2: generate_architecture_doc()
Tools-->>Gemini: Returns Markdown Blueprint & Milestones
Gemini-->>API: Synthesized Response + JSON Payload
API-->>Developer: Complete Project Blueprint
Talent Navigator Agent — Starter Codebase
Google I/O Extended Silicon Mountain Starter Kit
Agentic AI
Gemini 1.5 Flash
FastAPI
Docker
Cloud Run
Python
Google Cloud
Note🚀 Google I/O Extended Companion Repository
- Workshop Guide: Building the Talent Navigator Agent with Gemini 1.5 & Cloud Run
- Target Platform: Google Cloud Run + Artifact Registry + Secret Manager
- Stack: Python 3.11, Google GenAI SDK (
gemini-1.5-flash), FastAPI, Docker
🏗️ Architecture Flow
📂 Codebase File Explorer
Browse and copy the complete files below:
"""
FastAPI Server for Talent Navigator Agent.
Provides RESTful APIs for developers and attendees to interact with the Agentic AI.
"""
import os
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
from agent import execute_agentic_loop
app = FastAPI(
title="Talent Navigator Agent API",
description="Google I/O Extended Silicon Mountain — Agentic AI with Gemini 1.5 Flash, FastAPI & Cloud Run",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
)
# Enable CORS for cross-origin browser requests during live demos
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class NavigateRequest(BaseModel):
prompt: str = Field(
...,
description="Description of your current skills and project interests.",
json_schema_extra={"example": "I know a bit of Flutter and Tailwind CSS, but I want to build something for civic tech in Buea."},
)
class ToolCallRecord(BaseModel):
tool_name: str
args: Dict[str, Any]
result: Optional[Dict[str, Any]] = None
class NavigateResponse(BaseModel):
status: str
model: str
prompt: str
tool_calls_executed: List[ToolCallRecord]
response_text: Optional[str] = None
mode: str
@app.get("/", tags=["Health"])
def root():
return {
"event": "Google I/O Extended Silicon Mountain",
"app": "Talent Navigator Agent",
"status": "online",
"docs": "/docs",
"endpoints": {
"health": "/health",
"navigate": "/api/navigate (POST)"
}
}
@app.get("/health", tags=["Health"])
def health_check():
has_api_key = bool(os.getenv("GEMINI_API_KEY"))
return {
"status": "healthy",
"gemini_api_configured": has_api_key,
"environment": os.getenv("ENVIRONMENT", "production")
}
@app.post("/api/navigate", response_model=NavigateResponse, tags=["Agent"])
def navigate_talent(request: NavigateRequest):
if not request.prompt or len(request.prompt.strip()) < 5:
raise HTTPException(
status_code=400,
detail="Please provide a valid prompt describing your skills and domain interest."
)
result = execute_agentic_loop(request.prompt)
return result
if __name__ == "__main__":
import uvicorn
port = int(os.environ.get("PORT", 8080))
host = os.environ.get("HOST", "0.0.0.0")
print(f"🚀 Starting Talent Navigator Agent on http://{host}:{port}")
uvicorn.run("main:app", host=host, port=port, reload=True)"""
Agent Core: Talent Navigator Agent powered by Gemini 1.5 Flash and Function Calling.
Uses the Google GenAI SDK to autonomously invoke domain tools and generate architectural blueprints.
"""
import os
from typing import Dict, Any, List
from dotenv import load_dotenv
from tools import search_tech_stack, generate_architecture_doc
load_dotenv()
# System prompt defining Persona and Operational Directives
SYSTEM_INSTRUCTION = """
You are the Silicon Mountain Talent Navigator Agent, an expert AI Solution Architect mentor
assisting software engineering students and early-career developers in Buea, Cameroon.
When a developer shares their current skills and project interests:
1. Always analyze their current skills and target domain.
2. FIRST, call the `search_tech_stack` tool to discover complementary technologies, backend requirements, and cloud infrastructure.
3. SECOND, synthesize the findings and call the `generate_architecture_doc` tool with a concrete, customized 4-milestone plan.
4. Conclude with an inspiring, actionable message tailored to building and deploying on Google Cloud.
"""
def execute_agentic_loop(user_prompt: str) -> Dict[str, Any]:
"""
Executes the Talent Navigator agentic loop with Gemini 1.5 Flash and Function Calling.
"""
api_key = os.getenv("GEMINI_API_KEY")
mock_mode = os.getenv("ENABLE_MOCK_MODE", "false").lower() in ["true", "1", "yes"]
# Graceful fallback mock if offline, in mock mode, or API key missing (Wi-Fi Contingency)
if mock_mode or not api_key or api_key.startswith("your_"):
return _execute_mock_agent(user_prompt)
try:
from google import genai
from google.genai import types
client = genai.Client(
api_key=api_key,
http_options=types.HttpOptions(timeout=15000),
)
tools_list = [search_tech_stack, generate_architecture_doc]
tool_map = {
"search_tech_stack": search_tech_stack,
"generate_architecture_doc": generate_architecture_doc,
}
response = client.models.generate_content(
model="gemini-1.5-flash",
contents=user_prompt,
config=types.GenerateContentConfig(
system_instruction=SYSTEM_INSTRUCTION,
tools=tools_list,
temperature=0.2,
automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True)
),
)
tool_calls_record = []
function_calls = response.function_calls or []
executed_results = {}
# If Gemini emitted tool calls, execute them deterministically
if function_calls:
for fc in function_calls:
tool_name = fc.name
tool_args = dict(fc.args) if fc.args else {}
if tool_name in tool_map:
try:
tool_result = tool_map[tool_name](**tool_args)
executed_results[tool_name] = tool_result
tool_calls_record.append({
"tool_name": tool_name,
"args": tool_args,
"result": tool_result,
})
except Exception as tool_err:
tool_calls_record.append({
"tool_name": tool_name,
"args": tool_args,
"error": str(tool_err),
})
# Synthesize final response text
response_text = None
if "generate_architecture_doc" in executed_results:
arch = executed_results["generate_architecture_doc"]
response_text = (
f"### Talent Navigator Architecture Generated\n\n"
f"{arch.get('markdown_specification', '')}"
)
elif hasattr(response, "text") and response.text:
response_text = response.text
else:
detected_domain = "civic tech"
if "search_tech_stack" in executed_results:
detected_domain = executed_results["search_tech_stack"].get("matched_domain", "Civic tech").lower()
project_title = f"{detected_domain.title()} Solution Portal"
milestones = [
"Phase 1: Build FastAPI microservice endpoints on local Docker environment.",
"Phase 2: Integrate Firebase Auth and Firestore for real-time services.",
"Phase 3: Connect Google Gemini 1.5 Flash API for automated intelligence.",
"Phase 4: Deploy to Google Cloud Run via Artifact Registry with Secret Manager."
]
arch_doc = generate_architecture_doc(
project_title=project_title,
domain=f"{detected_domain.title()} in Buea Ecosystem",
frontend_stack="Flutter Mobile App + Tailwind Web",
backend_stack="FastAPI Microservice (Python)",
database_choice="Cloud SQL PostgreSQL or Firebase Firestore",
cloud_services=["Google Cloud Run", "Secret Manager", "Artifact Registry"],
milestones=milestones,
)
response_text = f"### Talent Navigator Architecture Generated\n\n{arch_doc['markdown_specification']}"
tool_calls_record.append({
"tool_name": "generate_architecture_doc",
"args": {"project_title": project_title, "domain": detected_domain},
"result": arch_doc,
})
return {
"status": "success",
"model": "gemini-1.5-flash",
"prompt": user_prompt,
"tool_calls_executed": tool_calls_record,
"response_text": response_text,
"mode": "live_gemini_api"
}
except Exception as e:
return {
"status": "partial_success",
"mode": "deterministic_fallback",
"error_note": f"API call failed ({str(e)}). Switched to deterministic agentic fallback.",
**_execute_mock_agent(user_prompt)
}
def _execute_mock_agent(user_prompt: str) -> Dict[str, Any]:
"""Deterministic fallback simulation for offline demos or unit testing."""
lower_prompt = user_prompt.lower()
detected_skills = []
for skill in ["flutter", "react", "tailwind", "python", "fastapi", "django", "node", "firebase", "sql"]:
if skill in lower_prompt:
detected_skills.append(skill.capitalize())
if not detected_skills:
detected_skills = ["Flutter", "Tailwind CSS"]
detected_domain = "civic tech"
for domain in ["civic tech", "fintech", "agritech", "healthtech", "edtech"]:
if domain in lower_prompt:
detected_domain = domain
break
stack_result = search_tech_stack(interest_domain=detected_domain, current_skills=detected_skills)
project_title = f"{detected_domain.title()} Citizen Portal"
milestones = [
"Phase 1: Build FastAPI microservice endpoints on local Docker environment.",
"Phase 2: Integrate Firebase Auth and Firestore for real-time citizen reports.",
"Phase 3: Connect Google Gemini 1.5 Flash API for automated issue classification.",
"Phase 4: Deploy to Google Cloud Run via Artifact Registry with Secret Manager."
]
arch_result = generate_architecture_doc(
project_title=project_title,
domain=f"{detected_domain.title()} in Buea Ecosystem",
frontend_stack=" + ".join(detected_skills) if detected_skills else "Flutter Mobile App",
backend_stack=stack_result["recommended_backend"],
database_choice=stack_result["recommended_database"],
cloud_services=[stack_result["recommended_cloud"]] + stack_result["essential_services"],
milestones=milestones,
)
return {
"status": "success",
"model": "gemini-1.5-flash (local agentic simulation)",
"prompt": user_prompt,
"tool_calls_executed": [
{"tool_name": "search_tech_stack", "args": {"interest_domain": detected_domain, "current_skills": detected_skills}, "result": stack_result},
{"tool_name": "generate_architecture_doc", "args": {"project_title": project_title, "domain": detected_domain}, "result": arch_result}
],
"response_text": (
f"### Talent Navigator Architecture Generated\n\n"
f"Based on your skills ({', '.join(detected_skills)}) and passion for **{detected_domain.title()}**, "
f"here is your complete production roadmap ready to build and deploy to Google Cloud Run.\n\n"
f"{arch_result['markdown_specification']}"
),
"mode": "deterministic_fallback"
}"""
Domain Tools for Talent Navigator Agent.
These functions represent the deterministic capabilities that the Gemini Agent can invoke.
"""
from typing import List, Dict, Any
# Local tech ecosystem knowledge base tailored for Silicon Mountain developers
DOMAIN_RECOMMENDATIONS = {
"civic tech": {
"recommended_backend": "FastAPI (Python) or Express (Node.js)",
"database": "Cloud SQL (PostgreSQL) or Firebase Firestore",
"cloud_infrastructure": "Google Cloud Run + Secret Manager",
"auth_and_services": ["Firebase Auth", "Cloud Storage for documents", "SMS/WhatsApp Alerts (Twilio/Infobip)"],
"recommended_additions": ["OpenStreetMap API", "GeoPandas", "Offline-first sync for low-connectivity regions"],
},
"fintech": {
"recommended_backend": "FastAPI or Go with strict typing",
"database": "PostgreSQL with ACID transaction compliance",
"cloud_infrastructure": "Google Cloud Run + Cloud KMS + Secret Manager",
"auth_and_services": ["MTN MoMo API / Orange Money API", "JWT Auth", "Cloud Audit Logs"],
"recommended_additions": ["Redis for rate-limiting", "Webhook idempotency handling", "Celery/Cloud Tasks for async payouts"],
},
"agritech": {
"recommended_backend": "FastAPI (Python) with ML inference pipeline",
"database": "Firebase Firestore (Offline caching) + Cloud Storage",
"cloud_infrastructure": "Google Cloud Run + Vertex AI / Gemini 1.5 Flash",
"auth_and_services": ["Firebase Auth", "USSD gateway or SMS notifications"],
"recommended_additions": ["Plant disease image classification", "Weather forecasting API", "Bilingual French/English UI"],
},
"healthtech": {
"recommended_backend": "FastAPI (Python)",
"database": "Cloud SQL PostgreSQL (Encrypted at rest)",
"cloud_infrastructure": "Google Cloud Run + Secret Manager",
"auth_and_services": ["Firebase Auth with MFA", "Encrypted Cloud Storage"],
"recommended_additions": ["Triage engine with Gemini", "Offline-first patient syncing"],
},
"edtech": {
"recommended_backend": "FastAPI or Django",
"database": "Firebase Firestore or Supabase",
"cloud_infrastructure": "Google Cloud Run + Cloud CDN",
"auth_and_services": ["Firebase Auth", "Cloud Storage for video/PDF materials"],
"recommended_additions": ["Gemini Flash AI tutor", "Lightweight bandwidth-saving mode"],
},
}
def search_tech_stack(interest_domain: str, current_skills: List[str]) -> Dict[str, Any]:
"""
Search and recommend complementary technologies, infrastructure, and services
based on the developer's target domain and their current skill set.
"""
normalized_domain = interest_domain.strip().lower()
matched_domain = "civic tech" # default fallback
for key in DOMAIN_RECOMMENDATIONS:
if key in normalized_domain:
matched_domain = key
break
profile = DOMAIN_RECOMMENDATIONS[matched_domain]
has_mobile_or_frontend = any(s.lower() in ["flutter", "react", "vue", "html", "tailwind", "css", "svelte", "android", "ios", "kotlin", "swift"] for s in current_skills)
has_backend = any(s.lower() in ["python", "fastapi", "django", "node", "express", "go", "java", "spring", "rust", "php", "laravel"] for s in current_skills)
stack_gaps = []
if not has_backend:
stack_gaps.append(f"Backend API layer needed -> Recommend {profile['recommended_backend']}")
if not has_mobile_or_frontend:
stack_gaps.append("Client UI needed -> Recommend Flutter (cross-platform mobile) or React (web)")
return {
"matched_domain": matched_domain.capitalize(),
"developer_current_skills": current_skills,
"recommended_backend": profile["recommended_backend"],
"recommended_database": profile["database"],
"recommended_cloud": profile["cloud_infrastructure"],
"essential_services": profile["auth_and_services"],
"specialized_additions": profile["recommended_additions"],
"identified_gaps": stack_gaps,
"deployment_target": "Google Cloud Run (Containerized Microservice)",
}
def generate_architecture_doc(
project_title: str,
domain: str,
frontend_stack: str,
backend_stack: str,
database_choice: str,
cloud_services: List[str],
milestones: List[str],
) -> Dict[str, Any]:
"""
Generate a concrete, production-grade project architecture specification
and step-by-step milestone roadmap for a software engineer or student project.
"""
services_rendered = "\n".join([f"- {s}" for s in cloud_services])
milestones_rendered = "\n".join([f"{i+1}. **{m}**" for i, m in enumerate(milestones)])
markdown_doc = f"""# {project_title} — Technical Architecture Specification
## 1. Executive Summary
- **Domain:** {domain}
- **Deployment Platform:** Google Cloud Run (Serverless Container)
- **Target Scale:** Scales from 0 to N instances on-demand
## 2. System Architecture
- **Client Application:** {frontend_stack}
- **API Microservice:** {backend_stack}
- **Persistence Layer:** {database_choice}
- **Cloud & Auxiliary Services:**
{services_rendered}
## 3. Implementation Roadmap
{milestones_rendered}
## 4. Google Cloud Deployment Checklist
1. Containerize the application using `Dockerfile` (optimized multi-stage build).
2. Authenticate to Google Artifact Registry.
3. Build and submit container image: `gcloud builds submit --tag europe-west1-docker.pkg.dev/$PROJECT_ID/io-extended-buea/{project_title.lower().replace(' ', '-')}:latest`.
4. Deploy to Cloud Run with Secret Manager: `gcloud run deploy --image ... --set-secrets=GEMINI_API_KEY=gemini-api-key:latest`.
"""
return {
"status": "success",
"project_title": project_title,
"domain": domain,
"architecture_summary": {
"frontend": frontend_stack,
"backend": backend_stack,
"database": database_choice,
"cloud_services": cloud_services,
},
"milestones": milestones,
"markdown_specification": markdown_doc.strip(),
}# Use lightweight official Python image
FROM python:3.11-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PORT=8080
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
curl \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port ${PORT:-8080}"]#!/bin/bash
set -e
PROJECT_ID=$(gcloud config get-value project 2>/dev/null)
REGION="europe-west1"
SERVICE_NAME="talent-navigator-agent"
REPO_NAME="io-extended-buea"
IMAGE_NAME="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${SERVICE_NAME}:latest"
echo "🚀 Deploying Talent Navigator Agent to Google Cloud Run..."
gcloud services enable \
run.googleapis.com \
artifactregistry.googleapis.com \
cloudbuild.googleapis.com \
secretmanager.googleapis.com
gcloud artifacts repositories create ${REPO_NAME} \
--repository-format=docker \
--location=${REGION} \
--description="Docker repo for I/O Extended" \
2>/dev/null || echo "Repository exists."
gcloud builds submit --tag ${IMAGE_NAME} .
# Ensure Secret Manager Access
if gcloud secrets describe gemini-api-key &>/dev/null; then
PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format='value(projectNumber)' 2>/dev/null || echo "")
if [ -n "$PROJECT_NUMBER" ]; then
gcloud secrets add-iam-policy-binding gemini-api-key \
--member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
--role="roles/secretmanager.secretAccessor" &>/dev/null || true
fi
gcloud run deploy ${SERVICE_NAME} \
--image ${IMAGE_NAME} \
--platform managed \
--region ${REGION} \
--allow-unauthenticated \
--set-secrets=GEMINI_API_KEY=gemini-api-key:latest \
--min-instances=0 \
--max-instances=5 \
--memory=512Mi \
--cpu=1
else
gcloud run deploy ${SERVICE_NAME} \
--image ${IMAGE_NAME} \
--platform managed \
--region ${REGION} \
--allow-unauthenticated \
--min-instances=0 \
--max-instances=5 \
--memory=512Mi \
--cpu=1
fi
SERVICE_URL=$(gcloud run services describe ${SERVICE_NAME} --platform managed --region ${REGION} --format 'value(status.url)')
echo "🎉 Deployed Successfully! Live URL: ${SERVICE_URL}"google-genai>=0.1.1
fastapi>=0.115.0
uvicorn[standard]>=0.30.0
pydantic>=2.8.0
python-dotenv>=1.0.1# Google Gemini API Key from Google AI Studio (https://aistudio.google.com/)
GEMINI_API_KEY=your_gemini_api_key_here
# Offline Simulation Mode (set to true to test locally without burning API quota)
ENABLE_MOCK_MODE=false
# Server Configuration
PORT=8080
HOST=0.0.0.0⚡ Quick Start Instructions
# 1. Clone or copy directory
cd code-snippets/talent-navigator-agent
# 2. Set up virtual environment
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
# 3. Configure .env with your Gemini API Key
cp .env.example .env
# 4. Launch FastAPI server
python main.pyOpen http://localhost:8080/docs to test the API directly.