Die Mission
Dauer: 2 Minuten
Du hast dich der Notfall-KI vorgestellt und dein Notsignal pulsiert jetzt auf der Planetenkarte – aber es ist kaum ein Flackern, das im Rauschen untergeht. Die Rettungsteams, die aus dem Orbit scannen, können etwas an Ihren Koordinaten sehen, aber sie können keine Verbindung herstellen. Das Signal ist zu schwach.
Damit das Signal des Beacons die volle Stärke erreicht, müssen Sie Ihren genauen Standort bestätigen. Das Navigationssystem der Kapsel ist ausgefallen, aber beim Absturz wurden verwertbare Beweise über die Absturzstelle verstreut. Bodenproben. Seltsame Flora. Freie Sicht auf den außerirdischen Nachthimmel.
Wenn Sie diese Beweise analysieren und feststellen können, in welcher Region der Erde Sie sich befinden, kann die KI Ihre Position triangulieren und das Beaconsignal verstärken. Vielleicht findet dich dann jemand.
Zeit, die Teile zusammenzusetzen.
Vorbereitung
⚠️ Für dieses Level ist der Abschluss von Level 0 erforderlich.
Prüfe vor dem Start, ob du Folgendes hast:
- [ ] config.json im Projektstammverzeichnis mit deiner Teilnehmer-ID und deinen Koordinaten
- [ ] Dein Avatar ist auf der Weltkarte sichtbar.
- [ ] Dein Beacon wird an deinen Koordinaten angezeigt (schwach).
Wenn Sie Stufe 0 noch nicht abgeschlossen haben, beginnen Sie damit.
Umfang
In diesem Level erstellen Sie ein KI-System mit mehreren Agenten, das Beweise vom Absturzort mithilfe der Parallelverarbeitung analysiert:
Lernziele
| Konzept | Lerninhalte |
|---|---|
| Multi-Agent-Systeme | Spezialisierte Agents mit einzelnen Verantwortlichkeiten erstellen |
| ParallelAgent | Unabhängige Agents erstellen, die gleichzeitig ausgeführt werden |
| before_agent_callback | Konfiguration abrufen und Status festlegen, bevor der Agent ausgeführt wird |
| ToolContext | Auf Statuswerte in Tool-Funktionen zugreifen |
| Benutzerdefinierte MCP-Server | Tools mit dem imperativen Muster erstellen (Python-Code in Cloud Run) |
| OneMCP BigQuery | Verbindung zum verwalteten MCP von Google für den BigQuery-Zugriff herstellen |
| Multimodale KI | Bilder und Video mit Audio mit Gemini analysieren |
| Agent-Orchestrierung | Mehrere Agents mit einem Root-Orchestrator koordinieren |
| Cloud-Bereitstellung | MCP-Server und ‑Agent in Cloud Run bereitstellen |
| Vorbereitung für A2A | Agents für die zukünftige Agent-zu-Agent-Kommunikation strukturieren |
Die Biome des Planeten
Die Oberfläche des Planeten ist in vier verschiedene Biome unterteilt, die jeweils einzigartige Merkmale aufweisen:
Anhand Ihrer Koordinaten wird bestimmt, in welchem Biom Sie abgestürzt sind. Die Beweise an der Absturzstelle spiegeln die Eigenschaften dieses Bioms wider:
| Biome | Quadrant | Geologische Beweise | Botanische Beweise | Astronomische Beweise |
|---|---|---|---|---|
| 🧊 CRYO | NW (x<50, y≥50) | Gefrorenes Methan, Eiskristalle | Frostfarne, Kryoflora | Blauer Riese |
| 🌋 VOLCANIC (Vulkanisch) | NE (x≥50, y≥50) | Obsidianablagerungen | Feuerblüten, hitzebeständige Flora | Doppelsternsystem mit rotem Zwerg |
| 💜 BIOLUMINESCENT | SW (x<50, y<50) | Phosphoreszierender Boden | Leuchtende Pilze, leuchtende Pflanzen | Grüner Pulsar |
| 🦴 FOSSILIZED | SE (x≥50, y<50) | Bernsteinvorkommen, Ite-Mineralien | Versteinerte Bäume, alte Flora | Gelbe Sonne |
Ihre Aufgabe: KI-Agents entwickeln, die die Beweise analysieren und ableiten können, in welchem Biom Sie sich befinden.
Umgebung einrichten
Dauer: 3 Minuten
Bevor Sie Nachweise generieren können, müssen Sie die erforderlichen Google Cloud APIs aktivieren, einschließlich OneMCP für BigQuery, das verwalteten MCP-Zugriff auf BigQuery bietet.
Umgebungseinrichtungsskript ausführen
👉💻 Führen Sie das Skript zur Einrichtung der Umgebung aus:
cd ~/way-back-home/level_1
chmod +x setup/setup_env.sh
./setup/setup_env.sh
Die Ausgabe sollte in etwa so aussehen:
================================================================
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!
================================================================
Quellumgebungsvariablen
👉💻 Umgebungsvariablen einlesen:
source ~/way-back-home/set_env.sh
Virtuelle Umgebung erstellen
👉💻 Erstellen und aktivieren Sie die virtuelle Python-Umgebung für Level 1:
cd ~/way-back-home/level_1
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
Sternenkatalog einrichten
👉💻 Sternenkatalog in BigQuery einrichten:
python setup/setup_star_catalog.py
Hier sollten Sie dies sehen:
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!
Beweise für Absturzstelle generieren
Dauer: 2 Minuten
Generieren Sie jetzt personalisierte Beweise für den Absturzort basierend auf Ihren Koordinaten.
Evidence Generator ausführen
👉💻 Führen Sie im Verzeichnis level_1 (mit aktiviertem venv) Folgendes aus:
cd ~/way-back-home/level_1
python generate_evidence.py
Die Ausgabe sollte in etwa so aussehen:
✓ 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!
==================================================
Nachweise prüfen
👉 Sehen Sie sich die generierten Beweisdateien im Ordner outputs/ an. Jeder spiegelt die Biomeigenschaften Ihres Absturzorts wider. Welches Biom es ist, erfahren Sie jedoch erst, wenn Ihre KI-Agents die Daten analysiert haben.
Je nach Standort kann der generierte Nachweis so aussehen:
Benutzerdefinierten MCP-Server erstellen
Dauer: 8 Minuten
Die Analysesysteme an Bord deiner Rettungskapsel sind ausgefallen, aber die Rohsensordaten haben den Absturz überstanden. Sie erstellen einen MCP-Server mit FastMCP, der Tools für geologische und botanische Analysen bereitstellt.
Geological Analysis Tool erstellen
Mit diesem Tool werden Bilder von Bodenproben analysiert, um die Mineralzusammensetzung zu ermitteln.
👉✏️ Öffnen Sie mcp-server/main.py und suchen Sie nach #REPLACE-GEOLOGICAL-TOOL. Ersetzen Sie sie durch:
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}
Botanisches Analysetool erstellen
Dieses Tool analysiert Videoaufnahmen von Pflanzen, einschließlich des Audiotracks.
👉✏️ Suchen Sie nach #REPLACE-BOTANICAL-TOOL und ersetzen Sie es durch:
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}
MCP-Server lokal testen
👉💻 MCP-Server testen:
cd ~/way-back-home/level_1/mcp-server
pip install -r requirements.txt
python main.py
Hier sollten Sie dies sehen:
[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
Der FastMCP-Server wird jetzt mit HTTP-Transport ausgeführt. Drücke Ctrl+C, um den Vorgang zu beenden.
MCP-Server in Cloud Run bereitstellen
👉💻 Bereitstellen:
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"
Dienst-URL speichern
👉💻 Dienst-URL speichern:
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
Spezialisten-Agents erstellen
Dauer: 8 Minuten
Als Nächstes erstellen Sie drei spezialisierte Agents, die jeweils für eine Aufgabe zuständig sind.
Geological Analyst Agent erstellen
👉✏️ Öffnen Sie agent/agents/geological_analyst.py und suchen Sie nach #REPLACE-GEOLOGICAL-AGENT. Ersetzen Sie sie durch:
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()]
)
Botanical Analyst-Agent erstellen
👉✏️ Öffnen Sie agent/agents/botanical_analyst.py und suchen Sie nach #REPLACE-BOTANICAL-AGENT. Ersetzen Sie sie durch:
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()]
)
Astronomical Analyst-Agent erstellen
Dieser Agent verwendet einen anderen Ansatz mit zwei Tool-Mustern:
- Lokales FunctionTool: Gemini Vision zum Extrahieren von Sternmerkmalen
- OneMCP BigQuery: Sternenkatalog über das verwaltete MCP von Google abfragen
👉✏️ Öffnen Sie agent/agents/astronomical_analyst.py und suchen Sie nach #REPLACE-ASTRONOMICAL-AGENT. Ersetzen Sie sie durch:
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]
)
MCP-Tool-Verbindungen erstellen
Dauer: 8 Minuten
Erstellen Sie die Tool-Wrapper, die eine Verbindung zu Ihrem bereitgestellten MCP-Server herstellen.
MCP-Tool-Verbindung erstellen (benutzerdefinierter MCP)
Dadurch wird eine Verbindung zu Ihrem benutzerdefinierten FastMCP-Server hergestellt, der in Cloud Run bereitgestellt wird.
👉✏️ Öffnen Sie agent/tools/mcp_tools.py und suchen Sie nach #REPLACE-MCP-TOOL-CONNECTION. Ersetzen Sie sie durch:
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()
Tools für die Sternanalyse erstellen (OneMCP BigQuery)
In diesem Abschnitt wird das verwaltete MCP-Muster veranschaulicht. Anstatt eigenen BigQuery-Clientcode zu schreiben, stellen wir eine Verbindung zum OneMCP-BigQuery-Server von Google her.
👉✏️ Öffnen Sie agent/tools/star_tools.py und suchen Sie nach #REPLACE-STAR-TOOLS. Ersetzen Sie sie durch:
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)
Orchestrator erstellen
Dauer: 8 Minuten
Erstellen Sie nun den Orchestrator für die parallele Crew und den Root-Orchestrator, der alles koordiniert.
Parallel Analysis Crew erstellen
Zuerst erstellen wir die Callback-Funktion und den ParallelAgent, der die Spezialisten gleichzeitig ausführt.
👉✏️ Öffnen Sie agent/agent.py und suchen Sie nach #REPLACE-PARALLEL-CREW. Ersetzen Sie sie durch:
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]
)
Root-Orchestrator erstellen
Erstellen Sie nun den Stamm-Agent, der alles koordiniert und den Callback verwendet.
👉✏️ Suchen Sie in derselben Datei (agent/agent.py) nach #REPLACE-ROOT-ORCHESTRATOR. Ersetzen Sie sie durch:
# =============================================================================
# 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
)
Standortbestätigungstool erstellen
Dieses Tool verwendet ToolContext, um Statuswerte zu lesen, die vom Callback festgelegt wurden.
👉✏️ Suche in agent/tools/confirm_tools.py nach #REPLACE-CONFIRM-TOOL. Ersetzen Sie sie durch:
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)
Mit der ADK-Web-UI testen
Dauer: 5 Minuten
Jetzt testen wir das gesamte Multi-Agent-System lokal.
ADK-Webserver starten
👉💻 Umgebungsvariablen festlegen und ADK-Webserver starten:
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
Hier sollten Sie dies sehen:
+-----------------------------------------------------------------------------+
| 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)
Auf die Web-UI zugreifen
👉 Wählen Sie in der Cloud Shell-Symbolleiste (rechts oben) über das Symbol Webvorschau die Option Port ändern aus.
👉 Legen Sie den Port auf 8000 fest und klicken Sie auf Ändern und Vorschau.
👉 Die ADK-Web-UI wird geöffnet. Wählen Sie im Drop-down-Menü agent aus.
Analyse ausführen
👉 Geben Sie in der Chatoberfläche Folgendes ein:
Analyze the evidence from my crash site and confirm my location to activate the beacon.
So funktioniert das Multi-Agenten-System:
- before_agent_callback wird zuerst ausgeführt und ruft Ihre Teilnehmerdaten ab.
- Der Root-Orchestrator empfängt Ihre Anfrage mit dem ausgefüllten Status.
- EvidenceAnalysisCrew wird aktiviert (ParallelAgent)
- Drei Spezialisten werden parallel ausgeführt mit
{key}-Vorlagen:- GeologicalAnalyst → sieht
{soil_url}, das aus dem Status aufgelöst wurde - BotanicalAnalyst → sieht
{flora_url}als behobenes Problem - AstronomicalAnalyst → sieht
{stars_url}und{project_id}als geklärt
- GeologicalAnalyst → sieht
- Root-Orchestrator synthetisiert (2 von 3)
- confirm_location aufgerufen mit ToolContext → „🔦 BEACON ACTIVATED!“
Im Trace-Bereich auf der rechten Seite werden alle Agent-Interaktionen und Tool-Aufrufe angezeigt.
👉 Drücken Sie im Terminal Ctrl+C, um den Server zu beenden, wenn Sie mit dem Testen fertig sind.
In Cloud Run bereitstellen
Dauer: 5 Minuten
Stellen Sie Ihr Multi-Agenten-System jetzt in Cloud Run bereit, um es für A2A vorzubereiten.
KI-Agenten bereitstellen
👉💻 Mit der ADK-CLI in Cloud Run bereitstellen:
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
Geben Sie bei Aufforderung Allow unauthenticated invocations to [mission-analysis-ai] (y/N)? y ein, um den öffentlichen Zugriff zu erlauben.
Die Ausgabe sollte in etwa so aussehen:
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!
Umgebungsvariablen in Cloud Run festlegen
Der bereitgestellte Agent muss auf Umgebungsvariablen zugreifen können. Aktualisieren Sie den Dienst:
👉💻 Legen Sie die erforderlichen Umgebungsvariablen fest:
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"
Agent-URL speichern
👉💻 Bereitgestellte URL abrufen:
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
Deployment prüfen
👉💻 Testen Sie den bereitgestellten Agenten, indem Sie die URL in Ihrem Browser öffnen (mit dem Flag --with_ui wurde die ADK-Weboberfläche bereitgestellt) oder indem Sie den folgenden curl-Befehl ausführen:
curl -X GET "$AGENT_URL/list-apps"
Sie sollten eine Antwort mit Ihrem Agenten sehen.
Fazit
Dauer: 1 Min.
Checkliste für die Bestätigung
✅ MCP-Server
- [ ] In Cloud Run bereitgestellt
- [ ] analyze_geological-Tool funktioniert
- [ ] analyze_botanical-Tool funktioniert
✅ Spezialisierte Agents
- [ ] GeologicalAnalyst verwendet {soil_url} aus dem Bundesstaat
- [ ] BotanicalAnalyst verwendet {flora_url} aus dem Bundesstaat
- [ ] AstronomicalAnalyst verwendet {stars_url} und {project_id} aus dem Bundesstaat
✅ before_agent_callback
- [ ] Ruft Teilnehmerdaten von der Backend-API ab
- [ ] Legt Statuswerte für alle untergeordneten Agents fest
- [ ] Funktioniert mit PARTICIPANT_ID aus der Umgebung
✅ ParallelAgent
- [ ] Alle drei Spezialisten werden gleichzeitig ausgeführt.
- [ ] Der Status wird über InvocationContext geteilt.
✅ Root Orchestrator
- [ ] Synthesizes with 2-of-3 agreement
- [ ] confirm_location uses ToolContext for state
- [ ] Beacon activates!
✅ Bereitstellung
- [ ] KI‑Agent in Cloud Run bereitgestellt
- [ ] A2A-Endpunkt zugänglich
✅ Weltkarte
- [ ] Beacon ist jetzt HELL (nicht dunkel)
- [ ] Biom wird beim Hovern angezeigt
🎉 Level 1 abgeschlossen!
Ihr Notsignal wird jetzt mit voller Stärke gesendet. Das triangulierte Signal durchdringt die atmosphärischen Störungen, ein stetiger Puls, der sagt: „Ich bin hier. Ich habe überlebt. Komm und such mich.“
Aber du bist nicht der Einzige auf diesem Planeten. Als dein Leuchtfeuer aktiviert wird, siehst du am Horizont andere Lichter flackern – andere Überlebende, andere Absturzstellen, andere Entdecker, die es geschafft haben.
In Stufe 2 lernen Sie, eingehende SOS-Signale zu verarbeiten und sich mit anderen Überlebenden abzustimmen. Bei der Rettung geht es nicht nur darum, gefunden zu werden, sondern auch darum, einander zu finden.
Fehlerbehebung
„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
„BigQuery-Tabelle nicht gefunden“
bash
python setup/setup_star_catalog.py
Spezialisten fragen nach URLs: Das bedeutet, dass die {key}-Vorlagen nicht funktionieren. Prüfen Sie, ob before_agent_callback für den Root-Agent festgelegt ist.
– Werden die Statuswerte durch die Callback-Einstellung korrekt festgelegt?
– Verwenden Sub-Agents {soil_url} (keine f-Strings)?
„Alle drei Analysen stimmen nicht überein“
Beweise neu generieren: python generate_evidence.py
„Agent not responding in adk web“ (Agent reagiert nicht im ADK-Web): - Prüfe, ob Port 8000 korrekt ist. - Prüfe, ob MCP_SERVER_URL und PARTICIPANT_ID festgelegt sind. - Prüfe das Terminal auf Fehlermeldungen.
Architekturübersicht
| Komponente | Typ | Muster | Zweck |
|---|---|---|---|
| setup_participant_context | Rückruf | before_agent_callback | Konfiguration abrufen, Status festlegen |
| GeologicalAnalyst | Agent | {soil_url}-Vorlagen | Bodenklassifizierung |
| BotanicalAnalyst | Agent | {flora_url}-Vorlagen | Klassifizierung von Pflanzen |
| AstronomicalAnalyst | Agent | {stars_url}, {project_id} | Stern-Triangulation |
| confirm_location | Tool | ToolContext-Statuszugriff | Beacon aktivieren |
| EvidenceAnalysisCrew | ParallelAgent | Zusammensetzung von Sub-Agents | Spezialisten gleichzeitig einsetzen |
| MissionAnalysisAI | Agent (Root) | Orchestrator + Callback | Koordinieren + synthetisieren |
| location-analyzer | FastMCP-Server | Benutzerdefiniertes MCP | Geologische und botanische Analyse |
| bigquery.googleapis.com/mcp | OneMCP | Verwaltetes MCP | BigQuery-Zugriff |
Beherrschte Schlüsselkonzepte
✅ before_agent_callback: Konfiguration vor dem Ausführen des KI-Agents abrufen
✅ {key} State Templating: Auf Statuswerte in KI-Agent-Anweisungen zugreifen
✅ ToolContext: Auf Statuswerte in Tool-Funktionen zugreifen
✅ State Sharing: Übergeordneter Status automatisch für untergeordnete KI-Agents über InvocationContext verfügbar
✅ Multi-Agent Architecture: Spezialisierte KI-Agents mit einzelnen Verantwortlichkeiten
✅ ParallelAgent: Gleichzeitige Ausführung unabhängiger Aufgaben
✅ Custom MCP Server: Eigener MCP-Server in Cloud Run
✅ OneMCP BigQuery: Verwaltetes MCP-Muster für den Datenbankzugriff
✅ Cloud Deployment: Statuslose Bereitstellung mit Umgebungsvariablen
✅ A2A Preparation: KI-Agent für die Kommunikation zwischen KI-Agents vorbereitet
Für Nicht-Gamer: Anwendungen in der realen Welt
„Pinpointing your location“ steht für Parallel Expert Analysis with Consensus (Parallele Expertenanalyse mit Konsens), bei der mehrere spezialisierte KI-Analysen gleichzeitig ausgeführt und die Ergebnisse zusammengeführt werden.
Unternehmensanwendungen
| Anwendungsfall | Parallel Experts | Syntheseregel |
|---|---|---|
| Medizinische Diagnose | Bildanalyst, Symptomanalyst, Laboranalyst | Konfidenzgrenzwert „2 von 3“ |
| Betrugserkennung | Transaktionsanalyst, Verhaltensanalyst, Netzwerkanalyst | 1 Flag = Überprüfung |
| Dokumentverarbeitung | OCR-Agent, Klassifizierungs-Agent, Extraktions-Agent | Alle müssen zustimmen |
| Qualitätskontrolle | Visueller Prüfer, Sensoranalyst, Spezifikationsprüfer | 2 von 3 bestanden |
Wichtige Erkenntnisse zur Architektur
before_agent_callback for configuration: Konfiguration einmal am Anfang abrufen, Status für alle untergeordneten Agents festlegen. Untergeordnete Agents lesen keine Konfigurationsdatei.
{key} State Templating: Deklarativ, übersichtlich, idiomatisch. Keine f-Strings, keine Importe, keine sys.path-Manipulation.
Konsensmechanismen: Eine 2‑von‑3-Vereinbarung kann Unklarheiten robust handhaben, ohne dass eine einstimmige Vereinbarung erforderlich ist.
ParallelAgent für unabhängige Aufgaben: Wenn Analysen nicht voneinander abhängen, können Sie sie gleichzeitig ausführen, um Zeit zu sparen.
Zwei MCP-Muster: benutzerdefiniert (selbst erstellt) und OneMCP (von Google gehostet). Beide verwenden StreamableHTTP.
Zustandslose Bereitstellung: Derselbe Code funktioniert lokal und bereitgestellt. Umgebungsvariablen + Backend-API = keine Konfigurationsdateien in Containern.
Wie geht es weiter?
Stufe 2: SOS-Signalverarbeitung →
Lernen Sie, eingehende Notsignale von anderen Überlebenden mithilfe ereignisgesteuerter Muster und einer erweiterten Agent-Koordination zu verarbeiten.