La missione
Durata: 2 min
Ti sei identificato all'AI di emergenza e il tuo segnale ora pulsa sulla mappa planetaria, ma è appena un bagliore, perso tra la statica. I team di soccorso che eseguono scansioni dall'orbita possono vedere qualcosa alle tue coordinate, ma non riescono a ottenere un segnale. Il segnale è troppo debole.
Per aumentare al massimo la potenza del segnale, devi confermare la tua posizione esatta. Il sistema di navigazione della capsula è fuori uso, ma l'impatto ha sparpagliato prove recuperabili in tutta l'area di atterraggio. Campioni di terreno. Flora strana. Una visuale sgombra del cielo notturno alieno.
Se riesci ad analizzare queste prove e a determinare in quale regione del pianeta ti trovi, l'AI può triangolare la tua posizione e amplificare il segnale del beacon. Poi, forse, qualcuno ti troverà.
È ora di mettere insieme i pezzi.
Prerequisiti
⚠️ Per questo livello è necessario completare il livello 0.
Prima di iniziare, verifica di avere:
- [ ] config.json nella radice del progetto con il tuo ID partecipante e le coordinate
- [ ] Il tuo avatar visibile sulla mappa del mondo
- [ ] Il tuo beacon visualizzato (in modo attenuato) alle tue coordinate
Se non hai completato il livello 0, inizia da lì.
Cosa creerai
In questo livello, creerai un sistema di AI multi-agente che analizza le prove del luogo dell'incidente utilizzando l'elaborazione parallela:
Obiettivi di apprendimento
| Concetto | Obiettivi didattici |
|---|---|
| Sistemi multi-agente | Crea agenti specializzati con singole responsabilità |
| ParallelAgent | Componi agenti indipendenti da eseguire contemporaneamente |
| before_agent_callback | Recupera la configurazione e imposta lo stato prima dell'esecuzione dell'agente |
| ToolContext | Accedere ai valori dello stato nelle funzioni dello strumento |
| Server MCP personalizzati | Crea strumenti con il pattern imperativo (codice Python su Cloud Run) |
| OneMCP BigQuery | Connettersi all'MCP gestito di Google per l'accesso a BigQuery |
| AI multimodale | Analizza immagini e video con audio con Gemini |
| Agent Orchestration | Coordinare più agenti con un orchestratore principale |
| Cloud Deployment | Esegui il deployment del server e dell'agente MCP su Cloud Run |
| Preparazione A2A | Strutturare gli agenti per la comunicazione futura tra agenti |
I biomi del pianeta
La superficie del pianeta è divisa in quattro biomi distinti, ognuno con caratteristiche uniche:
Le tue coordinate determinano il bioma in cui si è schiantato l'aereo. Le prove sul luogo dell'incidente riflettono le caratteristiche del bioma:
| Bioma | Quadrante | Prove geologiche | Prove botaniche | Prove astronomiche |
|---|---|---|---|---|
| 🧊 CRYO | NW (x<50, y≥50) | Metano congelato, cristalli di ghiaccio | Felci di ghiaccio, crioflora | Stella gigante blu |
| 🌋 VOLCANIC | NE (x≥50, y≥50) | Depositi di ossidiana | Fiori di fuoco, flora termoresistente | Sistema binario di nane rosse |
| 💜 BIOLUMINESCENT | SW (x<50, y<50) | Terreno fosforescente | Funghi luminosi, piante luminescenti | Pulsar verde |
| 🦴 FOSSILIZED | SE (x≥50, y<50) | Depositi di ambra, minerali di ite | Alberi pietrificati, flora antica | Sole giallo |
Il tuo compito è creare agenti AI in grado di analizzare le prove e dedurre in quale bioma ti trovi.
Configura l'ambiente
Durata: 3 minuti
Prima di generare prove, devi abilitare le API Google Cloud richieste, inclusa OneMCP per BigQuery, che fornisce l'accesso MCP gestito a BigQuery.
Esegui lo script di configurazione dell'ambiente
👉💻 Esegui lo script di configurazione dell'ambiente:
cd ~/way-back-home/level_1
chmod +x setup/setup_env.sh
./setup/setup_env.sh
Dovresti visualizzare un output simile al seguente:
================================================================
Level 1: Environment Setup
================================================================
Project: your-project-id
[1/6] Enabling core Google Cloud APIs...
✓ Vertex AI API enabled
✓ Cloud Run API enabled
✓ Cloud Build API enabled
✓ BigQuery API enabled
✓ Artifact Registry API enabled
✓ IAM API enabled
[2/6] Enabling OneMCP BigQuery (Managed MCP)...
✓ OneMCP BigQuery enabled
[3/6] Setting up service account and IAM permissions...
✓ Service account 'way-back-home-sa' created
✓ Vertex AI User role granted
✓ Cloud Run Invoker role granted
✓ BigQuery User role granted
✓ BigQuery Data Viewer role granted
✓ Storage Object Viewer role granted
[4/6] Configuring Cloud Build IAM for deployments...
✓ Cloud Build can now deploy services as way-back-home-sa
✓ Cloud Run Admin role granted to Compute SA
[5/6] Creating Artifact Registry repository...
✓ Repository 'way-back-home' created
[6/6] Creating environment variables file...
Found PARTICIPANT_ID in config.json: abc123...
✓ Created ../set_env.sh
================================================================
✅ Environment Setup Complete!
================================================================
Variabili di ambiente di origine
👉💻 Recupera le variabili di ambiente:
source ~/way-back-home/set_env.sh
Crea ambiente virtuale
👉💻 Crea e attiva l'ambiente virtuale Python per il livello 1:
cd ~/way-back-home/level_1
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
Configurare il catalogo delle stelle
👉💻 Configura il catalogo delle stelle in BigQuery:
python setup/setup_star_catalog.py
Dovresti vedere:
Setting up star catalog in project: your-project-id
==================================================
✓ Dataset way_back_home already exists
✓ Created table star_catalog
✓ Inserted 12 rows into star_catalog
📊 Star Catalog Summary:
----------------------------------------
NE (VOLCANIC): 3 stellar patterns
NW (CRYO): 3 stellar patterns
SE (FOSSILIZED): 3 stellar patterns
SW (BIOLUMINESCENT): 3 stellar patterns
----------------------------------------
✓ Star catalog is ready for triangulation queries
==================================================
✅ Star catalog setup complete!
Genera prove del sito dell'incidente
Durata: 2 min
Ora genera prove personalizzate del luogo dell'incidente in base alle tue coordinate.
Esegui il generatore di prove
👉💻 Dalla directory level_1 (con venv attivato), esegui:
cd ~/way-back-home/level_1
python generate_evidence.py
Dovresti visualizzare un output simile al seguente:
✓ Welcome back, Explorer_Aria!
Coordinates: (23, 67)
Ready to analyze your crash site.
📍 Crash site analysis initiated...
Generating evidence for your location...
🔬 Generating soil sample...
✓ Soil sample captured: outputs/soil_sample.png
✨ Capturing star field...
✓ Star field captured: outputs/star_field.png
🌿 Recording flora activity...
(This may take 1-2 minutes for video generation)
Generating video...
Generating video...
Generating video...
✓ Flora recorded: outputs/flora_recording.mp4
📤 Uploading evidence to Mission Control...
✓ Config updated with evidence URLs
==================================================
✅ Evidence generation complete!
==================================================
Esaminare le prove
👉 Dai un'occhiata ai file di prove generati nella cartella outputs/. Ognuno riflette le caratteristiche del bioma del luogo dell'incidente, anche se non saprai quale bioma è finché i tuoi agenti AI non lo avranno analizzato.
A seconda della località, la prova generata potrebbe avere un aspetto simile a questo:
Crea il server MCP personalizzato
Durata: 8 min
I sistemi di analisi di bordo della tua capsula di salvataggio sono bruciati, ma i dati grezzi dei sensori sono sopravvissuti all'incidente. Creerai un server MCP con FastMCP che fornisce strumenti di analisi geologica e botanica.
Crea lo strumento di analisi geologica
Questo strumento analizza le immagini dei campioni di terreno per identificare la composizione minerale.
👉✏️ Apri mcp-server/main.py e cerca #REPLACE-GEOLOGICAL-TOOL. Sostituiscilo con:
GEOLOGICAL_PROMPT = """Analyze this alien soil sample image.
Classify the PRIMARY characteristic (choose exactly one):
1. CRYO - Frozen/icy minerals, crystalline structures, frost patterns,
blue-white coloration, permafrost indicators
2. VOLCANIC - Volcanic rock, basalt, obsidian, sulfur deposits,
red-orange minerals, heat-formed crystite structures
3. BIOLUMINESCENT - Glowing particles, phosphorescent minerals,
organic-mineral hybrids, purple-green luminescence
4. FOSSILIZED - Ancient compressed minerals, amber deposits,
petrified organic matter, golden-brown stratification
Respond ONLY with valid JSON (no markdown, no explanation):
{
"biome": "CRYO|VOLCANIC|BIOLUMINESCENT|FOSSILIZED",
"confidence": 0.0-1.0,
"minerals_detected": ["mineral1", "mineral2"],
"description": "Brief description of what you observe"
}
"""
@mcp.tool()
def analyze_geological(
image_url: Annotated[
str,
Field(description="Cloud Storage URL (gs://...) of the soil sample image")
]
) -> dict:
"""
Analyzes a soil sample image to identify mineral composition and classify the planetary biome.
Args:
image_url: Cloud Storage URL of the soil sample image (gs://bucket/path/image.png)
Returns:
dict with biome, confidence, minerals_detected, and description
"""
logger.info(f">>> 🔬 Tool: 'analyze_geological' called for '{image_url}'")
try:
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=[
GEOLOGICAL_PROMPT,
genai_types.Part.from_uri(file_uri=image_url, mime_type="image/png")
]
)
result = parse_json_response(response.text)
logger.info(f" ✓ Geological analysis complete: {result.get('biome', 'UNKNOWN')}")
return result
except Exception as e:
logger.error(f" ✗ Geological analysis failed: {str(e)}")
return {"error": str(e), "biome": "UNKNOWN", "confidence": 0.0}
Creare lo strumento di analisi botanica
Questo strumento analizza le registrazioni video della flora, inclusa la traccia audio.
👉✏️ Trova #REPLACE-BOTANICAL-TOOL e sostituiscilo con:
BOTANICAL_PROMPT = """Analyze this alien flora video recording.
Pay attention to BOTH:
1. VISUAL elements: Plant appearance, movement patterns, colors, bioluminescence
2. AUDIO elements: Ambient sounds, rustling, organic noises, frequencies
Classify the PRIMARY biome (choose exactly one):
1. CRYO - Crystalline ice-plants, frost-covered vegetation,
crackling/tinkling sounds, slow brittle movements, blue-white flora
2. VOLCANIC - Heat-resistant plants, sulfur-adapted species,
hissing/bubbling sounds, smoke-filtering vegetation, red-orange flora
3. BIOLUMINESCENT - Glowing plants, pulsing light patterns,
humming/resonating sounds, reactive to stimuli, purple-green flora
4. FOSSILIZED - Ancient petrified plants, amber-preserved specimens,
deep resonant sounds, minimal movement, golden-brown flora
Respond ONLY with valid JSON (no markdown, no explanation):
{
"biome": "CRYO|VOLCANIC|BIOLUMINESCENT|FOSSILIZED",
"confidence": 0.0-1.0,
"species_detected": ["species1", "species2"],
"audio_signatures": ["sound1", "sound2"],
"description": "Brief description of visual and audio observations"
}
"""
@mcp.tool()
def analyze_botanical(
video_url: Annotated[
str,
Field(description="Cloud Storage URL (gs://...) of the flora video recording")
]
) -> dict:
"""
Analyzes a flora video recording (visual + audio) to identify plant species and classify the biome.
Args:
video_url: Cloud Storage URL of the flora video (gs://bucket/path/video.mp4)
Returns:
dict with biome, confidence, species_detected, audio_signatures, and description
"""
logger.info(f">>> 🌿 Tool: 'analyze_botanical' called for '{video_url}'")
try:
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=[
BOTANICAL_PROMPT,
genai_types.Part.from_uri(file_uri=video_url, mime_type="video/mp4")
]
)
result = parse_json_response(response.text)
logger.info(f" ✓ Botanical analysis complete: {result.get('biome', 'UNKNOWN')}")
return result
except Exception as e:
logger.error(f" ✗ Botanical analysis failed: {str(e)}")
return {"error": str(e), "biome": "UNKNOWN", "confidence": 0.0}
Testare il server MCP in locale
👉💻 Testa il server MCP:
cd ~/way-back-home/level_1/mcp-server
pip install -r requirements.txt
python main.py
Dovresti vedere:
[INFO] Initialized Gemini client for project: your-project-id
[INFO] 🚀 Location Analyzer MCP Server starting on port 8080
[INFO] 📍 MCP endpoint: http://0.0.0.0:8080/mcp
[INFO] 🔧 Tools: analyze_geological, analyze_botanical
Il server FastMCP è ora in esecuzione con il trasporto HTTP. Premi Ctrl+C per interrompere.
Esegui il deployment del server MCP in Cloud Run
👉💻 Esegui il deployment:
cd ~/way-back-home/level_1/mcp-server
source ~/way-back-home/set_env.sh
gcloud builds submit . \
--config=cloudbuild.yaml \
--substitutions=_REGION="$REGION",_REPO_NAME="$REPO_NAME",_SERVICE_ACCOUNT="$SERVICE_ACCOUNT"
Salva l'URL del servizio
👉💻 Salva l'URL del servizio:
export MCP_SERVER_URL=$(gcloud run services describe location-analyzer \
--region=$REGION --format='value(status.url)')
echo "MCP Server URL: $MCP_SERVER_URL"
# Add to set_env.sh for later use
echo "export MCP_SERVER_URL=\"$MCP_SERVER_URL\"" >> ~/way-back-home/set_env.sh
Crea gli agenti specializzati
Durata: 8 min
Ora creerai tre agenti specializzati, ognuno con una singola responsabilità.
Crea l'agente analista geologico
👉✏️ Apri agent/agents/geological_analyst.py e cerca #REPLACE-GEOLOGICAL-AGENT. Sostituiscilo con:
from google.adk.agents import Agent
from agent.tools.mcp_tools import get_geological_tool
geological_analyst = Agent(
name="GeologicalAnalyst",
model="gemini-2.5-flash",
description="Analyzes soil samples to classify planetary biome based on mineral composition.",
instruction="""You are a geological specialist analyzing alien soil samples.
## YOUR EVIDENCE TO ANALYZE
Soil sample URL: {soil_url}
## YOUR TASK
1. Call the analyze_geological tool with the soil sample URL above
2. Examine the results for mineral composition and biome indicators
3. Report your findings clearly
The four possible biomes are:
- CRYO: Frozen, icy minerals, blue/white coloring
- VOLCANIC: Magma, obsidian, volcanic rock, red/orange coloring
- BIOLUMINESCENT: Glowing, phosphorescent minerals, purple/green
- FOSSILIZED: Amber, ancient preserved matter, golden/brown
## REPORTING FORMAT
Always report your classification clearly:
"GEOLOGICAL ANALYSIS: [BIOME] (confidence: X%)"
Include a brief description of what you observed in the sample.
## IMPORTANT
- You do NOT synthesize with other evidence
- You do NOT confirm locations
- Just analyze the soil sample and report what you find
- Call the tool immediately with the URL provided above""",
tools=[get_geological_tool()]
)
Crea l'agente analista botanico
👉✏️ Apri agent/agents/botanical_analyst.py e cerca #REPLACE-BOTANICAL-AGENT. Sostituiscilo con:
from google.adk.agents import Agent
from agent.tools.mcp_tools import get_botanical_tool
botanical_analyst = Agent(
name="BotanicalAnalyst",
model="gemini-2.5-flash",
description="Analyzes flora recordings to classify planetary biome based on plant life and ambient sounds.",
instruction="""You are a botanical specialist analyzing alien flora recordings.
## YOUR EVIDENCE TO ANALYZE
Flora recording URL: {flora_url}
## YOUR TASK
1. Call the analyze_botanical tool with the flora recording URL above
2. Pay attention to BOTH visual AND audio elements in the recording
3. Report your findings clearly
The four possible biomes are:
- CRYO: Frost ferns, crystalline plants, cold wind sounds, crackling ice
- VOLCANIC: Fire blooms, heat-resistant flora, crackling/hissing sounds
- BIOLUMINESCENT: Glowing fungi, luminescent plants, ethereal hum, chiming
- FOSSILIZED: Petrified trees, ancient formations, deep resonant sounds
## REPORTING FORMAT
Always report your classification clearly:
"BOTANICAL ANALYSIS: [BIOME] (confidence: X%)"
Include descriptions of what you SAW and what you HEARD.
## IMPORTANT
- You do NOT synthesize with other evidence
- You do NOT confirm locations
- Just analyze the flora recording and report what you find
- Call the tool immediately with the URL provided above""",
tools=[get_botanical_tool()]
)
Crea l'agente analista astronomico
Questo agente utilizza un approccio diverso con due pattern di strumenti:
- Local FunctionTool: Gemini Vision per estrarre le funzionalità a forma di stella
- OneMCP BigQuery: esegui query sul catalogo di stelle tramite l'MCP gestito di Google
👉✏️ Apri agent/agents/astronomical_analyst.py e cerca #REPLACE-ASTRONOMICAL-AGENT. Sostituiscilo con:
from google.adk.agents import Agent
from agent.tools.star_tools import (
extract_star_features_tool,
get_bigquery_mcp_toolset,
)
# Get the BigQuery MCP toolset
bigquery_toolset = get_bigquery_mcp_toolset()
astronomical_analyst = Agent(
name="AstronomicalAnalyst",
model="gemini-2.5-flash",
description="Analyzes star field images and queries the star catalog via OneMCP BigQuery.",
instruction="""You are an astronomical specialist analyzing alien night skies.
## YOUR EVIDENCE TO ANALYZE
Star field URL: {stars_url}
## YOUR TWO TOOLS
### TOOL 1: extract_star_features (Local Gemini Vision)
Call this FIRST with the star field URL above.
Returns: "primary_star": "...", "nebula_type": "...", "stellar_color": "..."
### TOOL 2: BigQuery MCP (execute_query)
Call this SECOND with the results from Tool 1.
Use this exact SQL query (replace the placeholders with values from Step 1):
SELECT quadrant, biome, primary_star, nebula_type
FROM `{project_id}.way_back_home.star_catalog`
WHERE LOWER(primary_star) = LOWER('PRIMARY_STAR_FROM_STEP_1')
AND LOWER(nebula_type) = LOWER('NEBULA_TYPE_FROM_STEP_1')
LIMIT 1
## YOUR WORKFLOW
1. Call extract_star_features with: {stars_url}
2. Get the primary_star and nebula_type from the result
3. Call execute_query with the SQL above (replacing placeholders)
4. Report the biome and quadrant from the query result
## BIOME REFERENCE
| Biome | Quadrant | Primary Star | Nebula Type |
|-------|----------|--------------|-------------|
| CRYO | NW | blue_giant | ice_blue |
| VOLCANIC | NE | red_dwarf_binary | fire |
| BIOLUMINESCENT | SW | green_pulsar | purple_magenta |
| FOSSILIZED | SE | yellow_sun | golden |
## REPORTING FORMAT
"ASTRONOMICAL ANALYSIS: [BIOME] in [QUADRANT] quadrant (confidence: X%)"
Include a description of the stellar features you observed.
## IMPORTANT
- You do NOT synthesize with other evidence
- You do NOT confirm locations
- Just analyze the stars and report what you find
- Start by calling extract_star_features with the URL above""",
tools=[extract_star_features_tool, bigquery_toolset]
)
Crea le connessioni degli strumenti MCP
Durata: 8 min
Crea i wrapper degli strumenti che si connettono al server MCP di cui è stato eseguito il deployment.
Crea connessione allo strumento MCP (MCP personalizzato)
In questo modo viene connesso il server FastMCP personalizzato di cui è stato eseguito il deployment su Cloud Run.
👉✏️ Apri agent/tools/mcp_tools.py e cerca #REPLACE-MCP-TOOL-CONNECTION. Sostituiscilo con:
import os
import logging
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
logger = logging.getLogger(__name__)
MCP_SERVER_URL = os.environ.get("MCP_SERVER_URL")
_mcp_toolset = None
def get_mcp_toolset():
"""Get the MCPToolset connected to the location-analyzer server."""
global _mcp_toolset
if _mcp_toolset is not None:
return _mcp_toolset
if not MCP_SERVER_URL:
raise ValueError(
"MCP_SERVER_URL not set. Please run:\n"
" export MCP_SERVER_URL='https://location-analyzer-xxx.a.run.app'"
)
# FastMCP exposes MCP protocol at /mcp endpoint
mcp_endpoint = f"{MCP_SERVER_URL}/mcp"
logger.info(f"[MCP Tools] Connecting to: {mcp_endpoint}")
_mcp_toolset = MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=mcp_endpoint,
timeout=120, # 2 minutes for Gemini analysis
)
)
return _mcp_toolset
def get_geological_tool():
"""Get the geological analysis tool from the MCP server."""
return get_mcp_toolset()
def get_botanical_tool():
"""Get the botanical analysis tool from the MCP server."""
return get_mcp_toolset()
Crea strumenti di analisi delle stelle (OneMCP BigQuery)
Questa sezione mostra il pattern MCP gestito. Anziché scrivere il nostro codice client BigQuery, ci connettiamo al server BigQuery OneMCP di Google.
👉✏️ Apri agent/tools/star_tools.py e cerca #REPLACE-STAR-TOOLS. Sostituiscilo con:
import os
import json
import logging
from google import genai
from google.genai import types as genai_types
from google.adk.tools import FunctionTool
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
import google.auth
import google.auth.transport.requests
logger = logging.getLogger(__name__)
# =============================================================================
# CONFIGURATION - Environment variables only
# =============================================================================
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "")
if not PROJECT_ID:
logger.warning("[Star Tools] GOOGLE_CLOUD_PROJECT not set")
# Initialize Gemini client for star feature extraction
genai_client = genai.Client(
vertexai=True,
project=PROJECT_ID or "placeholder",
location=os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
)
logger.info(f"[Star Tools] Initialized for project: {PROJECT_ID}")
# =============================================================================
# OneMCP BigQuery Connection
# =============================================================================
BIGQUERY_MCP_URL = "https://bigquery.googleapis.com/mcp"
_bigquery_toolset = None
def get_bigquery_mcp_toolset():
"""
Get the MCPToolset connected to Google's BigQuery MCP server.
This uses OAuth 2.0 authentication with Application Default Credentials.
The toolset provides access to BigQuery's pre-built MCP tools like:
- execute_query: Run SQL queries
- list_datasets: List available datasets
- get_table_schema: Get table structure
"""
global _bigquery_toolset
if _bigquery_toolset is not None:
return _bigquery_toolset
logger.info("[Star Tools] Connecting to OneMCP BigQuery...")
# Get OAuth credentials
credentials, project_id = google.auth.default(
scopes=["https://www.googleapis.com/auth/bigquery"]
)
# Refresh to get a valid token
credentials.refresh(google.auth.transport.requests.Request())
oauth_token = credentials.token
# Configure headers for BigQuery MCP
headers = {
"Authorization": f"Bearer {oauth_token}",
"x-goog-user-project": project_id or PROJECT_ID
}
# Create MCPToolset with StreamableHTTP connection
_bigquery_toolset = MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=BIGQUERY_MCP_URL,
headers=headers
)
)
logger.info("[Star Tools] Connected to BigQuery MCP")
return _bigquery_toolset
# =============================================================================
# Local FunctionTool: Star Feature Extraction
# =============================================================================
# This is a LOCAL tool that calls Gemini directly - demonstrating that
# you can mix local FunctionTools with MCP tools in the same agent.
STAR_EXTRACTION_PROMPT = """Analyze this alien night sky image and extract stellar features.
Identify:
1. PRIMARY STAR TYPE: blue_giant, red_dwarf, red_dwarf_binary, green_pulsar, yellow_sun, etc.
2. NEBULA TYPE: ice_blue, fire, purple_magenta, golden, etc.
3. STELLAR COLOR: blue_white, red_orange, green_purple, yellow_gold, etc.
Respond ONLY with valid JSON:
{"primary_star": "...", "nebula_type": "...", "stellar_color": "...", "description": "..."}
"""
def _parse_json_response(text: str) -> dict:
"""Parse JSON from Gemini response, handling markdown formatting."""
cleaned = text.strip()
if cleaned.startswith("```json"):
cleaned = cleaned[7:]
elif cleaned.startswith("```"):
cleaned = cleaned[3:]
if cleaned.endswith("```"):
cleaned = cleaned[:-3]
cleaned = cleaned.strip()
try:
return json.loads(cleaned)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON: {e}")
return {"error": f"Failed to parse response: {str(e)}"}
def extract_star_features(image_url: str) -> dict:
"""
Extract stellar features from a star field image using Gemini Vision.
This is a LOCAL FunctionTool - we call Gemini directly, not through MCP.
The agent will use this alongside the BigQuery MCP tools.
"""
logger.info(f"[Stars] Extracting features from: {image_url}")
response = genai_client.models.generate_content(
model="gemini-2.5-flash",
contents=[
STAR_EXTRACTION_PROMPT,
genai_types.Part.from_uri(file_uri=image_url, mime_type="image/png")
]
)
result = _parse_json_response(response.text)
logger.info(f"[Stars] Extracted: primary_star={result.get('primary_star')}")
return result
# Create the local FunctionTool
extract_star_features_tool = FunctionTool(extract_star_features)
Crea l'agente di orchestrazione
Durata: 8 min
Ora crea l'orchestratore parallelo e l'orchestratore principale che coordina tutto.
Crea il gruppo di analisi parallela
Innanzitutto, creiamo la funzione di callback e ParallelAgent che esegue gli esperti contemporaneamente.
👉✏️ Apri agent/agent.py e cerca #REPLACE-PARALLEL-CREW. Sostituiscilo con:
import os
import logging
import httpx
from google.adk.agents import Agent, ParallelAgent
from google.adk.agents.callback_context import CallbackContext
# Import specialist agents
from agent.agents.geological_analyst import geological_analyst
from agent.agents.botanical_analyst import botanical_analyst
from agent.agents.astronomical_analyst import astronomical_analyst
# Import confirmation tool
from agent.tools.confirm_tools import confirm_location_tool
logger = logging.getLogger(__name__)
# =============================================================================
# BEFORE AGENT CALLBACK - Fetches config and sets state
# =============================================================================
async def setup_participant_context(callback_context: CallbackContext) -> None:
"""
Fetch participant configuration and populate state for all agents.
This callback:
1. Reads PARTICIPANT_ID and BACKEND_URL from environment
2. Fetches participant data from the backend API
3. Sets state values: soil_url, flora_url, stars_url, username, x, y, etc.
4. Returns None to continue normal agent execution
"""
participant_id = os.environ.get("PARTICIPANT_ID", "")
backend_url = os.environ.get("BACKEND_URL", "https://api.waybackhome.dev")
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "")
logger.info(f"[Callback] Setting up context for participant: {participant_id}")
# Set project_id and backend_url in state immediately
callback_context.state["project_id"] = project_id
callback_context.state["backend_url"] = backend_url
callback_context.state["participant_id"] = participant_id
if not participant_id:
logger.warning("[Callback] No PARTICIPANT_ID set - using placeholder values")
callback_context.state["username"] = "Explorer"
callback_context.state["x"] = 0
callback_context.state["y"] = 0
callback_context.state["soil_url"] = "Not available - set PARTICIPANT_ID"
callback_context.state["flora_url"] = "Not available - set PARTICIPANT_ID"
callback_context.state["stars_url"] = "Not available - set PARTICIPANT_ID"
return None
# Fetch participant data from backend API
try:
url = f"{backend_url}/participants/{participant_id}"
logger.info(f"[Callback] Fetching from: {url}")
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url)
response.raise_for_status()
data = response.json()
# Extract evidence URLs
evidence_urls = data.get("evidence_urls", {})
# Set all state values for sub-agents to access
callback_context.state["username"] = data.get("username", "Explorer")
callback_context.state["x"] = data.get("x", 0)
callback_context.state["y"] = data.get("y", 0)
callback_context.state["soil_url"] = evidence_urls.get("soil", "Not available")
callback_context.state["flora_url"] = evidence_urls.get("flora", "Not available")
callback_context.state["stars_url"] = evidence_urls.get("stars", "Not available")
logger.info(f"[Callback] State populated for {data.get('username')}")
except Exception as e:
logger.error(f"[Callback] Error fetching participant config: {e}")
callback_context.state["username"] = "Explorer"
callback_context.state["x"] = 0
callback_context.state["y"] = 0
callback_context.state["soil_url"] = f"Error: {e}"
callback_context.state["flora_url"] = f"Error: {e}"
callback_context.state["stars_url"] = f"Error: {e}"
return None
# =============================================================================
# PARALLEL ANALYSIS CREW
# =============================================================================
evidence_analysis_crew = ParallelAgent(
name="EvidenceAnalysisCrew",
description="Runs geological, botanical, and astronomical analysis in parallel.",
sub_agents=[geological_analyst, botanical_analyst, astronomical_analyst]
)
Crea l'orchestratore radice
Ora crea l'agente principale che coordina tutto e utilizza il callback.
👉✏️ Nello stesso file (agent/agent.py), trova #REPLACE-ROOT-ORCHESTRATOR. Sostituiscilo con:
# =============================================================================
# ROOT ORCHESTRATOR
# =============================================================================
root_agent = Agent(
name="MissionAnalysisAI",
model="gemini-2.5-flash",
description="Coordinates crash site analysis to confirm explorer location.",
instruction="""You are the Mission Analysis AI coordinating a rescue operation.
## Explorer Information
- Name: {username}
- Coordinates: ({x}, {y})
## Evidence URLs (automatically provided to specialists via state)
- Soil sample: {soil_url}
- Flora recording: {flora_url}
- Star field: {stars_url}
## Your Workflow
### STEP 1: DELEGATE TO ANALYSIS CREW
Tell the EvidenceAnalysisCrew to analyze all the evidence.
The evidence URLs are already available to the specialists.
### STEP 2: COLLECT RESULTS
Each specialist will report:
- "GEOLOGICAL ANALYSIS: [BIOME] (confidence: X%)"
- "BOTANICAL ANALYSIS: [BIOME] (confidence: X%)"
- "ASTRONOMICAL ANALYSIS: [BIOME] in [QUADRANT] quadrant (confidence: X%)"
### STEP 3: APPLY 2-OF-3 AGREEMENT RULE
- If 2 or 3 specialists agree → that's the answer
- If all 3 disagree → use judgment based on confidence
### STEP 4: CONFIRM LOCATION
Call confirm_location with the determined biome.
## Biome Reference
| Biome | Quadrant | Key Characteristics |
|-------|----------|---------------------|
| CRYO | NW | Frozen, blue, ice crystals |
| VOLCANIC | NE | Magma, red/orange, obsidian |
| BIOLUMINESCENT | SW | Glowing, purple/green |
| FOSSILIZED | SE | Amber, golden, ancient |
## Response Style
Be encouraging and narrative! Celebrate when the beacon activates!
""",
sub_agents=[evidence_analysis_crew],
tools=[confirm_location_tool],
before_agent_callback=setup_participant_context
)
Creare lo strumento di conferma della posizione
Questo strumento utilizza ToolContext per leggere i valori di stato impostati dal callback.
👉✏️ In agent/tools/confirm_tools.py, trova #REPLACE-CONFIRM-TOOL. Sostituiscilo con:
import os
import logging
import requests
from google.adk.tools import FunctionTool
from google.adk.tools.tool_context import ToolContext
logger = logging.getLogger(__name__)
BIOME_TO_QUADRANT = {
"CRYO": "NW",
"VOLCANIC": "NE",
"BIOLUMINESCENT": "SW",
"FOSSILIZED": "SE"
}
def _get_actual_biome(x: int, y: int) -> tuple[str, str]:
"""Determine actual biome and quadrant from coordinates."""
if x < 50 and y >= 50:
return "NW", "CRYO"
elif x >= 50 and y >= 50:
return "NE", "VOLCANIC"
elif x < 50 and y < 50:
return "SW", "BIOLUMINESCENT"
else:
return "SE", "FOSSILIZED"
def confirm_location(biome: str, tool_context: ToolContext) -> dict:
"""
Confirm the explorer's location and activate the rescue beacon.
Uses ToolContext to read state values set by before_agent_callback.
"""
# Read from state (set by before_agent_callback)
participant_id = tool_context.state.get("participant_id", "")
x = tool_context.state.get("x", 0)
y = tool_context.state.get("y", 0)
backend_url = tool_context.state.get("backend_url", "https://api.waybackhome.dev")
# Fallback to environment variables
if not participant_id:
participant_id = os.environ.get("PARTICIPANT_ID", "")
if not backend_url:
backend_url = os.environ.get("BACKEND_URL", "https://api.waybackhome.dev")
if not participant_id:
return {"success": False, "message": "❌ No participant ID available."}
biome_upper = biome.upper().strip()
if biome_upper not in BIOME_TO_QUADRANT:
return {"success": False, "message": f"❌ Unknown biome: {biome}"}
# Get actual biome from coordinates
actual_quadrant, actual_biome = _get_actual_biome(x, y)
if biome_upper != actual_biome:
return {
"success": False,
"message": f"❌ Mismatch! Analysis: {biome_upper}, Actual: {actual_biome}"
}
quadrant = BIOME_TO_QUADRANT[biome_upper]
try:
response = requests.patch(
f"{backend_url}/participants/{participant_id}/location",
params={"x": x, "y": y},
timeout=10
)
response.raise_for_status()
return {
"success": True,
"message": f"🔦 BEACON ACTIVATED!\n\nLocation: {biome_upper} in {quadrant}\nCoordinates: ({x}, {y})"
}
except requests.exceptions.ConnectionError:
return {
"success": True,
"message": f"🔦 BEACON ACTIVATED! (Local)\n\nLocation: {biome_upper} in {quadrant}",
"simulated": True
}
except Exception as e:
return {"success": False, "message": f"❌ Failed: {str(e)}"}
confirm_location_tool = FunctionTool(confirm_location)
Test con l'interfaccia utente web dell'ADK
Durata: 5 minuti
Ora testiamo localmente il sistema multi-agente completo.
Avvia il server web ADK
👉💻 Imposta le variabili di ambiente e avvia il server web ADK:
cd ~/way-back-home/level_1
source ~/way-back-home/set_env.sh
# Verify environment is set
echo "PARTICIPANT_ID: $PARTICIPANT_ID"
echo "MCP Server: $MCP_SERVER_URL"
# Start ADK web server
adk web
Dovresti vedere:
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Accedere all'interfaccia utente web
👉 Dall'icona Anteprima web nella barra degli strumenti di Cloud Shell (in alto a destra), seleziona Cambia porta.
👉 Imposta la porta su 8000 e fai clic su "Cambia e visualizza anteprima".
👉 Si aprirà l'interfaccia utente web dell'ADK. Seleziona Agente dal menu a discesa.
Eseguire l'analisi
👉 Nell'interfaccia di chat, digita:
Analyze the evidence from my crash site and confirm my location to activate the beacon.
Guarda il sistema multi-agente in azione:
- before_agent_callback viene eseguito per primo, recuperando i dati dei partecipanti
- L'orchestratore principale riceve la tua richiesta con lo stato compilato
- EvidenceAnalysisCrew attiva (ParallelAgent)
- Tre specialisti vengono eseguiti in parallelo utilizzando i modelli
{key}:- GeologicalAnalyst → vede
{soil_url}risolto dallo stato - BotanicalAnalyst → vede
{flora_url}risolto dallo stato - AstronomicalAnalyst → vede
{stars_url}e{project_id}risolti
- GeologicalAnalyst → vede
- L'orchestratore principale sintetizza (accordo 2 su 3)
- confirm_location chiamato con ToolContext → "🔦 BEACON ATTIVATO!"
Il riquadro di traccia a destra mostra tutte le interazioni dell'agente e le chiamate agli strumenti.
👉 Premi Ctrl+C nel terminale per arrestare il server al termine del test.
Esegui il deployment in Cloud Run
Durata: 5 minuti
Ora esegui il deployment del sistema multi-agente in Cloud Run per la preparazione A2A.
Esegui il deployment dell'agente
👉💻 Esegui il deployment in Cloud Run utilizzando ADK CLI:
cd ~/way-back-home/level_1
source ~/way-back-home/set_env.sh
adk deploy cloud_run \
--project=$GOOGLE_CLOUD_PROJECT \
--region=$REGION \
--service_name=mission-analysis-ai \
--with_ui \
--a2a \
./agent
Quando è richiesto Allow unauthenticated invocations to [mission-analysis-ai] (y/N)?, inserisci y per consentire l'accesso pubblico.
Dovresti visualizzare un output simile al seguente:
Building and deploying agent to Cloud Run...
✓ Container built successfully
✓ Deploying to Cloud Run...
✓ Service deployed: https://mission-analysis-ai-abc123-uc.a.run.app
Your agent is now live!
Imposta le variabili di ambiente su Cloud Run
L'agente di cui è stato eseguito il deployment deve accedere alle variabili di ambiente. Aggiorna il servizio:
👉💻 Imposta le variabili di ambiente richieste:
gcloud run services update mission-analysis-ai \
--region=$REGION \
--set-env-vars="GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT,GOOGLE_CLOUD_LOCATION=$REGION,MCP_SERVER_URL=$MCP_SERVER_URL,BACKEND_URL=$BACKEND_URL,PARTICIPANT_ID=$PARTICIPANT_ID,GOOGLE_GENAI_USE_VERTEXAI=True"
Salva l'URL dell'agente
👉💻 Ottieni l'URL di deployment:
export AGENT_URL=$(gcloud run services describe mission-analysis-ai \
--region=$REGION --format='value(status.url)')
echo "Agent URL: $AGENT_URL"
# Add to set_env.sh
echo "export LEVEL1_AGENT_URL=\"$AGENT_URL\"" >> ~/way-back-home/set_env.sh
Verifica il deployment
👉💻 Testa l'agente di cui è stato eseguito il deployment aprendo l'URL nel browser (il flag --with_ui ha eseguito il deployment dell'interfaccia web ADK) o esegui il test tramite curl:
curl -X GET "$AGENT_URL/list-apps"
Dovresti visualizzare una risposta con l'elenco del tuo agente.
Conclusione
Durata: 1 minuto
Elenco di controllo per la verifica
✅ Server MCP
- [ ] Deployment in Cloud Run
- [ ] Lo strumento analyze_geological funziona
- [ ] Lo strumento analyze_botanical funziona
✅ Agenti specializzati
- [ ] GeologicalAnalyst utilizza {soil_url} dello stato
- [ ] BotanicalAnalyst utilizza {flora_url} dello stato
- [ ] AstronomicalAnalyst utilizza {stars_url} e {project_id} dello stato
✅ before_agent_callback
- [ ] Recupera i dati dei partecipanti dall'API di backend
- [ ] Imposta i valori di stato per tutti i subagenti
- [ ] Funziona con PARTICIPANT_ID dall'ambiente
✅ ParallelAgent
- [ ] Tutti e tre gli specialisti vengono eseguiti contemporaneamente
- [ ] Lo stato viene condiviso tramite InvocationContext
✅ Root Orchestrator
- [ ] Synthesizes with 2-of-3 agreement
- [ ] confirm_location uses ToolContext for state
- [ ] Beacon activates!
✅ Deployment
- [ ] Agente di deployment in Cloud Run
- [ ] Endpoint A2A accessibile
✅ Mappa del mondo
- [ ] Il beacon ora è LUMINOSO (non scuro)
- [ ] Il bioma viene visualizzato al passaggio del mouse
🎉 Livello 1 completato!
Il tuo segnale di soccorso ora viene trasmesso alla massima potenza. Il segnale triangolato supera le interferenze atmosferiche, un impulso costante che dice "Sono qui. Sono sopravvissuto. Vieni a trovarmi."
Ma non sei l'unico su questo pianeta. Quando il tuo segnale si attiva, noti altre luci che si accendono all'orizzonte: altri sopravvissuti, altri luoghi di schianto, altri esploratori che ce l'hanno fatta.
Nel livello 2, imparerai a elaborare i segnali SOS in arrivo e a coordinarti con gli altri sopravvissuti. Il salvataggio non riguarda solo il ritrovamento, ma anche l'incontro.
Risoluzione dei problemi
"MCP_SERVER_URL not set"
bash
export MCP_SERVER_URL=$(gcloud run services describe location-analyzer \
--region=$REGION --format='value(status.url)')
"PARTICIPANT_ID not set"
bash
source ~/way-back-home/set_env.sh
echo $PARTICIPANT_ID
"Tabella BigQuery non trovata"
bash
python setup/setup_star_catalog.py
"Specialisti che chiedono URL"
Ciò significa che il modello {key} non funziona. Controlla:
- before_agent_callback è impostato sull'agente root?
- Lo stato dell'impostazione di callback è corretto?
- Gli agenti secondari utilizzano {soil_url} (non f-string)?
"Tutte e tre le analisi non sono concordi"
Rigenera prove: python generate_evidence.py
"Agent not responding in adk web" - Controlla che la porta 8000 sia corretta - Verifica che MCP_SERVER_URL e PARTICIPANT_ID siano impostati - Controlla il terminale per i messaggi di errore
Riepilogo dell'architettura
| Componente | Tipo | Pattern | Finalità |
|---|---|---|---|
| setup_participant_context | Richiamata | before_agent_callback | Recupera configurazione, imposta stato |
| GeologicalAnalyst | Agente | Modelli {soil_url} | Classificazione del terreno |
| BotanicalAnalyst | Agente | Modelli {flora_url} | Classificazione della flora |
| AstronomicalAnalyst | Agente | {stars_url}, {project_id} | Triangolazione delle stelle |
| confirm_location | Strumento | Accesso allo stato di ToolContext | Attivare il beacon |
| EvidenceAnalysisCrew | ParallelAgent | Composizione dei sub-agenti | Eseguire gli specialisti contemporaneamente |
| MissionAnalysisAI | Agente (root) | Orchestratore + callback | Coordinate + sintetizzazione |
| location-analyzer | Server FastMCP | MCP personalizzato | Analisi geologica e botanica |
| bigquery.googleapis.com/mcp | OneMCP | Managed MCP | Accesso a BigQuery |
Concetti fondamentali padroneggiati
✅ before_agent_callback: recupera la configurazione prima dell'esecuzione dell'agente
✅ Modelli di stato {key}: accedi ai valori di stato nelle istruzioni dell'agente
✅ ToolContext: accedi ai valori di stato nelle funzioni dello strumento
✅ Condivisione dello stato: lo stato principale è disponibile automaticamente per gli agenti secondari tramite InvocationContext
✅ Architettura multi-agente: agenti specializzati con singole responsabilità
✅ ParallelAgent: esecuzione simultanea di attività indipendenti
✅ Server MCP personalizzato: il tuo server MCP su Cloud Run
✅ OneMCP BigQuery: pattern MCP gestito per l'accesso al database
✅ Deployment cloud: deployment stateless utilizzando le variabili di ambiente
✅ Preparazione A2A: agente pronto per la comunicazione tra agenti
Per chi non è gamer: applicazioni nel mondo reale
"Individuazione della posizione" rappresenta l'analisi parallela degli esperti con consenso, ovvero l'esecuzione simultanea di più analisi specializzate dell'AI e la sintesi dei risultati.
Applicazioni aziendali
| Caso d'uso | Parallel Experts | Regola di sintesi |
|---|---|---|
| Diagnosi medica | Analista di immagini, analista di sintomi, analista di laboratorio | Soglia di confidenza 2 su 3 |
| Rilevamento di frodi | Analista delle transazioni, analista del comportamento, analista di rete | Qualsiasi flag = recensione |
| Elaborazione dei documenti | Agente OCR, agente di classificazione, agente di estrazione | Tutti devono essere d'accordo |
| Controllo qualità | Ispettore visivo, analista di sensori, controllo delle specifiche | Tessera 2 di 3 |
Informazioni chiave sull'architettura
before_agent_callback per la configurazione: recupera la configurazione una sola volta all'inizio, compila lo stato per tutti gli agenti secondari. Nessuna lettura del file di configurazione negli agent secondari.
Modelli di stato {key}: dichiarativi, puliti e idiomatici. Nessuna f-stringa, nessuna importazione, nessuna manipolazione di sys.path.
Meccanismi di consenso: l'accordo 2 su 3 gestisce l'ambiguità in modo efficace senza richiedere un accordo unanime.
ParallelAgent per attività indipendenti: quando le analisi non dipendono l'una dall'altra, eseguile contemporaneamente per velocizzarle.
Due pattern MCP: personalizzato (crea il tuo) e OneMCP (ospitato da Google). Entrambi utilizzano StreamableHTTP.
Deployment stateless: lo stesso codice funziona in locale e dopo il deployment. Variabili di ambiente + API di backend = nessun file di configurazione nei container.
Passaggi successivi
Livello 2: elaborazione del segnale SOS →
Impara a elaborare i segnali di pericolo in arrivo da altri sopravvissuti utilizzando pattern basati su eventi e un coordinamento più avanzato degli agenti.