Way Back Home - Level 1: Pinpoint Location

1. المهمة

لقد عرّفت عن نفسك للذكاء الاصطناعي الخاص بخدمات الطوارئ، وبدأ جهاز التتبُّع الخاص بك الآن بإرسال إشارات نبضية على الخريطة الكوكبية، ولكنها بالكاد تظهر، إذ إنّها غير واضحة وسط التشويش. يمكن لفرق الإنقاذ التي تجري عمليات المسح من المدار رؤية شيء ما في إحداثياتك، ولكن لا يمكنها تحديد موقعك بدقة. الإشارة ضعيفة جدًا.

لتعزيز قوة إشارة جهاز التتبّع، عليك تأكيد موقعك الجغرافي الدقيق. تعطّل نظام التنقّل في المركبة، ولكن تسبّب الحادث في انتشار أدلة يمكن استردادها في موقع الهبوط. عيّنات التربة نباتات غريبة منظر واضح لسماء الليل الغريبة

إذا كان بإمكانك تحليل هذا الدليل وتحديد المنطقة الجغرافية التي تتواجد فيها على الكوكب، يمكن للذكاء الاصطناعي تحديد موقعك الجغرافي بدقة وتضخيم إشارة جهاز التتبّع. بعد ذلك، قد يعثر عليك أحد المستخدمين.

حان الوقت لربط كل هذه المعلومات ببعضها.

المتطلبات الأساسية

⚠️ يتطلّب هذا المستوى إكمال المستوى 0.

قبل البدء، تأكَّد من توفّر ما يلي:

• [ ] config.json في جذر المشروع مع رقم تعريف المشارك والإحداثيات
• [ ] ظهور الأفاتار الخاص بك على خريطة العالم
• [ ] ظهور جهاز الإرسال (بشكل خافت) عند إحداثياتك

إذا لم تكمل المستوى 0، ابدأ به أولاً.

ما ستنشئه

في هذا المستوى، ستنشئ نظام ذكاء اصطناعي متعدد الوكلاء يحلّل أدلة موقع التحطّم باستخدام المعالجة المتوازية:

ملاحظة من المهندس المعماري: نمطان من MCP

يعلّم هذا المستوى نمطَين أساسيَّين من أنماط MCP:

1. MCP مخصّص (نمط إلزامي): يمكنك إنشاء خادم MCP الخاص بك على Cloud Run باستخدام أدوات مخصّصة. تكتب رمز Python البرمجي الذي ينفّذ كل أداة. تُستخدم هنا للتحليل الجيولوجي والنباتي باستخدام Gemini Vision.
2. OneMCP (النمط المُدار): يمكنك الربط بخوادم OneMCP التي أنشأتها Google مسبقًا (مثل bigquery.googleapis.com/mcp). الأدوات متوفّرة، ما عليك سوى ضبط عملية الربط. تُستخدَم هنا لطلبات البحث في فهرس النجوم.

ويعكس ذلك قرارات التصميم في العالم الحقيقي:

• المعالجة المخصّصة: عندما تحتاج إلى منطق مخصّص للذكاء الاصطناعي أو معالجة خاصة بمجال معيّن أو خوارزميات خاصة
• OneMCP: عندما تحتاج إلى وصول عادي إلى قاعدة البيانات أو عمليات دمج لواجهات برمجة التطبيقات أو الوصول إلى خدمات Google

أهداف التعلّم

الفكرة	ما ستتعلمه
الأنظمة المتعددة الوكلاء	إنشاء وكلاء متخصصين بمسؤوليات فردية
ParallelAgent	إنشاء وكلاء مستقلين لتشغيلهم في الوقت نفسه
before_agent_callback	استرجاع الإعدادات وضبط الحالة قبل تشغيل البرنامج
ToolContext	الوصول إلى قيم الحالة في وظائف الأدوات
خوادم MCP المخصّصة	إنشاء أدوات باستخدام النمط الإلزامي (رمز Python البرمجي على Cloud Run)
OneMCP BigQuery	الربط بمنصة MCP المُدارة من Google للوصول إلى BigQuery
الذكاء الاصطناعي المتعدّد الوسائط	تحليل الصور والفيديوهات والصوت باستخدام Gemini
Agent Orchestration	تنسيق عمل وكلاء متعدّدين باستخدام أداة تنسيق رئيسية
النشر على السحابة الإلكترونية	نشر خادم MCP والوكيل على Cloud Run
التحضير لعملية الربط بين التطبيقين	بنية الوكلاء للتواصل المستقبلي بين الوكلاء

المناطق الأحيائية على الكوكب

ينقسم سطح الكوكب إلى أربع مناطق أحيائية مميزة، ولكل منها خصائص فريدة:

تحدّد إحداثياتك المنطقة الأحيائية التي تحطّمت فيها المركبة. تعكس الأدلة في موقع التحطّم خصائص المنطقة الأحيائية:

المنطقة الأحيائية	الربع	الأدلة الجيولوجية	الأدلة النباتية	الأدلة الفلكية
🧊 CRYO	الشمال الغربي (x<50, y≥50)	الميثان المتجمد، بلورات الجليد	نباتات الصقيع، نباتات التجميد	نجم عملاق أزرق
🌋 VOLCANIC	NE (x≥50, y≥50)	رواسب الأوبسيديان والأوبسيديانيت	زهور النار، نباتات مقاومة للحرارة	نظام ثنائي من الأقزام الحمراء
💜 BIOLUMINESCENT	SW (x<50, y<50)	تربة فسفورية	الفطريات المتوهجة والنباتات المضيئة	نجم نابض أخضر
🦴 متحجّر	SE (x≥50, y<50)	رواسب الكهرمان ومعادن الإيت	أشجار متحجرة ونباتات قديمة	شمس صفراء

مهمتك هي إنشاء وكلاء ذكاء اصطناعي يمكنهم تحليل الأدلة واستنتاج المنطقة الأحيائية التي تتواجد فيها.

2. إعداد البيئة

هل تحتاج إلى أرصدة Google Cloud؟

• إذا كنت تحضر ورشة عمل بإشراف معلّم: سيقدّم لك المعلّم رمز رصيد. يُرجى استخدام الرمز الذي يقدّمونه.
• إذا كنت تعمل على هذا الدرس التطبيقي بنفسك: يمكنك الاستفادة من رصيد مجاني على Google Cloud لتغطية تكاليف ورشة العمل. يُرجى النقر على هذا الرابط للحصول على رصيد واتّباع الخطوات الواردة في دليل الفيديو أدناه لتطبيقه على حسابك.

تشغيل نص برمجي لإعداد البيئة

قبل إنشاء المستندات الثبوتية، عليك تفعيل واجهات برمجة التطبيقات المطلوبة في Google Cloud، بما في ذلك OneMCP لخدمة BigQuery التي توفّر إمكانية الوصول المُدار إلى BigQuery من خلال MCP.

ملاحظة من المهندس المعماري: OneMCP

‫OneMCP هي خدمة MCP مُدارة من Google. بدلاً من إنشاء خادم MCP خاص بك للوصول إلى BigQuery، يمكنك الربط بخادم Google المُنشأ مسبقًا على bigquery.googleapis.com/mcp. يوضّح هذا المثال نمط MCP المُدار، حيث يمكنك ضبط الاتصال وتوفّر Google الأدوات.

👉💻 شغِّل نص إعداد البيئة البرمجي:

cd $HOME/way-back-home/level_1
chmod +x setup/setup_env.sh
./setup/setup_env.sh

من المفترَض أن تظهر لك نتيجة مثل:

================================================================
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!
================================================================

متغيّرات البيئة المصدر

👉💻 الحصول على متغيرات البيئة:

source $HOME/way-back-home/set_env.sh

تثبيت الحِزم التابعة

👉💻 ثبِّت تبعيات المستوى 1 في Python:

cd $HOME/way-back-home/level_1
uv sync

إعداد كتالوج النجوم

👉💻 إعداد كتالوج النجوم في BigQuery:

uv run python setup/setup_star_catalog.py

سيظهر لك ما يلي:

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!

3- Generate Crash Site Evidence

يمكنك الآن إنشاء دليل مخصّص على موقع الحادث استنادًا إلى إحداثياتك.

ملاحظة من المهندس المعماري: إنشاء المحتوى باستخدام الذكاء الاصطناعي

توضّح هذه الخطوة عملية إنشاء المحتوى باستخدام الذكاء الاصطناعي. ينشئ Gemini صورًا وفيديوهات للأدلة مخصّصة لموقعك الجغرافي المحدّد على الكرة الأرضية. ويتم استخدام هذه التقنية نفسها في مرحلة الإنتاج لإنشاء بيانات تدريب مخصّصة، وأدوات اختبار، ومحتوى اصطناعي.

تشغيل أداة "إنشاء المستندات الداعمة"

👉💻 من الدليل level_1، نفِّذ ما يلي:

cd $HOME/way-back-home/level_1
uv run python generate_evidence.py

من المفترَض أن تظهر لك نتيجة مثل:

✓ 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!
==================================================

مراجعة الدليل

👉 خذ لحظة لإلقاء نظرة على ملفات المستندات التي تم إنشاؤها في المجلد outputs/. يعكس كل منها خصائص المنطقة الأحيائية لموقع الحادث، ولكن لن تعرف المنطقة الأحيائية إلا بعد أن يحلّلها وكلاء الذكاء الاصطناعي.

قد تبدو المستندات الثبوتية التي تم إنشاؤها على النحو التالي حسب موقعك الجغرافي:

4. إنشاء خادم MCP مخصّص

لقد تعطلت أنظمة التحليل المضمّنة في كبسولة النجاة، ولكن بيانات الاستشعار الأولية نجت من التحطّم. ستنشئ خادم MCP باستخدام FastMCP يوفّر أدوات تحليل جيولوجي ونباتي.

ملاحظة من المهندس المعماري: FastMCP

نحن بصدد إنشاء خادم MCP مخصّص باستخدام FastMCP، وهو إطار عمل Pythonic يتعامل مع جميع تفاصيل بروتوكول MCP. ما عليك سوى تحديد الأدوات باستخدام أداة التزيين @mcp.tool(). يستخدم الخادم بروتوكول HTTP للنقل عند نشر Cloud Run، ما يؤدي إلى عرض بروتوكول MCP على /mcp.

إنشاء "أداة التحليل الجيولوجي"

تحلّل هذه الأداة صور عيّنات التربة لتحديد التركيب المعدني.

👉✏️ افتح $HOME/way-back-home/level_1/mcp-server/main.py وابحث عن #REPLACE-GEOLOGICAL-TOOL. استبدِلها بما يلي:

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}

إنشاء أداة التحليل النباتي

تحلّل هذه الأداة تسجيلات الفيديو الخاصة بالنباتات، بما في ذلك المقطع الصوتي.

👉✏️ في الملف نفسه ($HOME/way-back-home/level_1/mcp-server/main.py)، ابحث عن #REPLACE-BOTANICAL-TOOL واستبدله بما يلي:

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}

ملاحظة

باستخدام FastMCP، لن تحتاج إلى تسجيل الأدوات يدويًا، إذ يتولّى الديكور @mcp.tool() كل شيء تلقائيًا.

اختبار خادم MCP محليًا

👉💻 اختبار خادم MCP:

cd $HOME/way-back-home/level_1/mcp-server
pip install -r requirements.txt
python main.py

سيظهر لك ما يلي:

[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

يعمل خادم FastMCP الآن مع بروتوكول نقل HTTP. اضغط على Ctrl+C للإيقاف.

نشر خادم MCP على Cloud Run

👉💻 النشر:

cd $HOME/way-back-home/level_1/mcp-server
source $HOME/way-back-home/set_env.sh

gcloud builds submit . \
  --config=cloudbuild.yaml \
  --substitutions=_REGION="$REGION",_REPO_NAME="$REPO_NAME",_SERVICE_ACCOUNT="$SERVICE_ACCOUNT"

حفظ عنوان URL للخدمة

👉💻 احفظ عنوان URL الخاص بالخدمة:

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\"" >> $HOME/way-back-home/set_env.sh

5- إنشاء وكلاء متخصصين

الآن، ستنشئ ثلاثة وكلاء متخصصين، لكل منهم مسؤولية واحدة.

ملاحظة من المهندس المعماري: مبدأ المسؤولية الفردية

هذا هو مبدأ المسؤولية الفردية المطبَّق على وكلاء الذكاء الاصطناعي. يجب أن يستوفي كل متخصص الشروط التالية:

• لديه وظيفة واحدة
• يستخدم أدوات محدّدة لهذه الوظيفة
• عرض نتيجة منظَّمة

ويسهّل ذلك اختبار الوكلاء وتصحيح أخطائهم واستبدالهم.

إنشاء وكيل محلّل جيولوجي

👉✏️ افتح agent/agents/geological_analyst.py وابحث عن #REPLACE-GEOLOGICAL-AGENT. استبدِلها بما يلي:

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()]
)

إنشاء وكيل محلّل النباتات

👉✏️ افتح agent/agents/botanical_analyst.py وابحث عن #REPLACE-BOTANICAL-AGENT. استبدِلها بما يلي:

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()]
)

إنشاء وكيل محلّل فلكي

يستخدم هذا الوكيل نهجًا مختلفًا مع نمطَي أدوات:

1. Local FunctionTool: استخدام "Gemini Vision" لاستخراج ميزات النجوم
2. OneMCP BigQuery: طلب البحث في كتالوج النجوم من خلال MCP المُدار من Google

ما هي OneMCP؟

OneMCP هي خدمة MCP مُدارة من Google توفّر خوادم MCP مُنشأة مسبقًا لخدمات Google Cloud. بدلاً من كتابة رمز برنامج العميل الخاص بك في BigQuery، يمكنك الربط بخادم Google المستضاف على bigquery.googleapis.com/mcp والوصول فورًا إلى أدوات مثل execute_query. يمكنك اعتبارها "BigQuery كأداة"، إذ لا يلزم إجراء أي عملية إعداد بخلاف تفعيل واجهة برمجة التطبيقات.

ملاحظة من المهندس المعماري: بنى الأدوات المختلطة

يوضّح ذلك كيفية دمج الأدوات المحلية مع أدوات MCP المُدارة. يستخدم الوكيل extract_star_features (أداة FunctionTool محلية) ثم يستعلم عن BigQuery من خلال أداة execute_query في OneMCP. يتم استبدال العنصرَين النائبَين {stars_url} و{project_id} في وقت التشغيل من الحالة.

👉✏️ افتح agent/agents/astronomical_analyst.py وابحث عن #REPLACE-ASTRONOMICAL-AGENT. استبدِلها بما يلي:

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]
)

6. إنشاء عمليات ربط بأداة MCP

ستنشئ الآن برامج تضمين Python التي تتيح لوكلاء ADK التفاعل مع خوادم MCP. تتعامل هذه البرامج المغلّفة مع دورة حياة الاتصال، أي إنشاء الجلسات واستدعاء الأدوات وتحليل الردود.

إنشاء اتصال بأداة MCP (أداة MCP مخصّصة)

يؤدي ذلك إلى الاتصال بخادم FastMCP المخصّص الذي تم نشره على Cloud Run.

ملاحظة من المهندس المعماري: ما هي فائدة StreamableHTTPConnectionParams؟

تتيح MCP استخدام طبقات نقل متعددة (stdio وSSE وHTTP). بما أنّ خادم MCP يتم نشره على Cloud Run كخدمة ويب، يجب أن يتوفّر نقل بيانات HTTP. يطلب StreamableHTTPConnectionParams من حزمة تطوير التطبيقات (ADK) التواصل مع خادم MCP عبر HTTP، وتحديدًا من خلال الاتصال بنقطة النهاية /mcp التي تعرضها FastMCP تلقائيًا. هذه هي وسيلة النقل العادية لخوادم MCP المستضافة على السحابة الإلكترونية.

👉✏️ افتح agent/tools/mcp_tools.py وابحث عن #REPLACE-MCP-TOOL-CONNECTION. استبدِلها بما يلي:

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()

إنشاء أدوات تحليل النجوم (OneMCP BigQuery)

يحتوي فهرس النجوم الذي حمّلته سابقًا إلى BigQuery على أنماط نجمية لكل منطقة حيوية. بدلاً من كتابة رمز برنامج BigQuery للعميل من أجل طلب البحث، نتصل بخادم OneMCP BigQuery من Google، والذي يعرض إمكانية execute_query في BigQuery كأداة MCP يمكن لأي وكيل ADK استخدامها مباشرةً.

ملاحظة من المهندس المعماري: نمطان MCP، أسلوب الربط نفسه

لاحظ الفرق:

• MCP مخصّص (القسم أعلاه): يمكنك كتابة رموز Python البرمجية باستخدام FastMCP ونشرها على Cloud Run واستخدام StreamableHTTP
• OneMCP (هذا القسم): تستضيف Google الخادم، ما عليك سوى ضبط OAuth والربط

تستخدم كلتا الخدمتين StreamableHTTPConnectionParams ولكنّهما تتصلان بخوادم مختلفة:

• المدة المخصّصة: https://your-cloud-run-service.a.run.app/mcp
• OneMCP: https://bigquery.googleapis.com/mcp

👉✏️ افتح agent/tools/star_tools.py وابحث عن #REPLACE-STAR-TOOLS. استبدِلها بما يلي:

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)

7. إنشاء أداة التنسيق

الآن، أنشئ طاقمًا متوازيًا ومنسّقًا رئيسيًا ينسّق كل شيء.

إنشاء فريق "التحليل الموازي"

لماذا يجب أن يعمل المتخصّصون الثلاثة بالتوازي؟ لأنّها مستقلة تمامًا، فالمحلّل الجيولوجي لا يحتاج إلى انتظار نتائج المحلّل النباتي، والعكس صحيح. يحلّل كلّ خبير جزءًا مختلفًا من الأدلة باستخدام أدوات مختلفة. يتم تشغيل ParallelAgent الثلاثة في الوقت نفسه، ما يقلّل إجمالي وقت التحليل من 30 ثانية تقريبًا (متسلسل) إلى 10 ثوانٍ تقريبًا (متوازٍ).

ما هي before_agent_callback؟

في "حزمة تطوير التطبيقات"، before_agent_callback هي دالة يتم تنفيذها قبل أن يبدأ الوكيل في معالجة أي طلب من المستخدم. يمكنك اعتبارها خطوة إعداد، أي "خطوة التحضير" التي يتّخذها الوكيل.

في ما يلي سبب أهمية ذلك لهذه البنية:

1. يتم تشغيله مرة واحدة عندما يبدأ الوكيل في معالجة الطلب
2. يستردّ بيانات المشاركين من واجهة برمجة التطبيقات الخلفية باستخدام PARTICIPANT_ID
3. تعبئة حالة الموظف بقيم مثل soil_url وflora_url وstars_url
4. تصل جميع الوكلاء الفرعيين تلقائيًا إلى هذه القيم من خلال {key} إنشاء النماذج في تعليماتهم

بدون رد الاتصال هذا، سيحتاج كل وكيل فرعي إلى معرفة كيفية جلب الإعدادات بنفسه. تنفّذ الدالة رد الاتصال هذا الإجراء مرة واحدة، وتوزّعه ميزة مشاركة الحالة في كل مكان. يمكنك الاطّلاع على المستندات الخاصة بـ ADK لمعرفة المزيد عن عمليات معاودة الاتصال.

وهذا يعني ما يلي:

• لا تتم قراءة ملف الإعداد في الوكلاء الفرعيين
• ما مِن عمليات استيراد لأدوات الإعداد
• يعمل الرمز نفسه محليًا وعلى Cloud Run
• طلب بيانات واحد من واجهة برمجة التطبيقات بدلاً من طلبات متعدّدة

👉✏️ افتح agent/agent.py وابحث عن #REPLACE-PARALLEL-CREW. استبدِلها بما يلي:

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]
)

إنشاء أداة التنسيق الرئيسية

الآن، أنشئ الوكيل الجذر الذي ينسّق كل شيء ويستخدم وظيفة معاودة الاتصال.

👉✏️ في الملف نفسه (agent/agent.py)، ابحث عن #REPLACE-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
)

إنشاء "أداة تأكيد الموقع الجغرافي"

هذه هي الخطوة الأخيرة، وهي الأداة التي تؤكّد موقعك الجغرافي لـ Mission Control وتفعّل جهاز التتبّع. عندما يحدّد المنسّق الرئيسي المنطقة الحيوية التي تتواجد فيها (باستخدام قاعدة الاتفاق بين اثنين من ثلاثة)، يستدعي هذه الأداة لإرسال النتيجة إلى واجهة برمجة التطبيقات الخلفية.

تستخدِم هذه الأداة ToolContext، ما يمنحها إذن الوصول إلى قيم الحالة (مثل participant_id وbackend_url) التي تم ضبطها بواسطة before_agent_callback سابقًا.

👉✏️ في agent/tools/confirm_tools.py، ابحث عن #REPLACE-CONFIRM-TOOL. استبدِلها بما يلي:

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)

ملاحظة من المهندس المعماري: ToolContext للوصول إلى الحالة

يمكن للأدوات الوصول إلى الحالة من خلال ToolContext:

def my_tool(arg: str, tool_context: ToolContext) -> dict: value = tool_context.state.get("key")

بهذه الطريقة، يحصل confirm_location على participant_id والإحداثيات وbackend_url التي تم ضبطها بواسطة ردّ الاتصال.

8. الاختبار باستخدام واجهة مستخدم ADK على الويب

لنختبر الآن نظام الوكلاء المتعدّدين الكامل على جهازك.

بدء خادم الويب الخاص بـ ADK

👉💻 ضبط متغيّرات البيئة وبدء خادم الويب الخاص بـ ADK:

cd $HOME/way-back-home/level_1
source $HOME/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
uv run adk web

سيظهر لك ما يلي:

+-----------------------------------------------------------------------------+
| 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)

الوصول إلى واجهة المستخدم على الويب

👉 من رمز معاينة الويب في شريط أدوات Cloud Shell (أعلى يسار الصفحة)، اختَر تغيير المنفذ.

👉 اضبط المنفذ على 8000 وانقر على "تغيير ومعاينة".

👉 سيتم فتح واجهة مستخدم ADK على الويب. اختَر الوكيل من القائمة المنسدلة.

تشغيل التحليل

👉 في واجهة المحادثة، اكتب:

Analyze the evidence from my crash site and confirm my location to activate the beacon.

شاهِد النظام المتعدّد الوكلاء أثناء العمل:

👉 عندما يكمل العملاء الثلاثة تحليلاتهم، اكتب:

Where am I?

طريقة معالجة النظام لطلبك:

تعرض لوحة التتبُّع على يسار الصفحة جميع تفاعلات الموظفين واستدعاءات الأدوات.

👉 اضغط على Ctrl+C في نافذة الوحدة الطرفية لإيقاف الخادم عند الانتهاء من الاختبار.

9- النشر على Cloud Run

يمكنك الآن نشر نظامك المتعدّد الوكلاء على Cloud Run ليكون جاهزًا لعمليات A2A.

ملاحظة من فريق التصميم: نشر حزمة تطوير التطبيقات (ADK)

يتعامل الأمر adk deploy cloud_run مع جميع تعقيدات النشر:

• إنشاء حاوية تتضمّن رمز الوكيل
• يتم النشر على Cloud Run مع الإعدادات المناسبة
• إعداد نقطة النهاية المتوافقة مع A2A تلقائيًا

نشر الوكيل

👉💻 النشر على Cloud Run باستخدام واجهة سطر الأوامر (CLI) الخاصة بـ ADK:

cd $HOME/way-back-home/level_1
source $HOME/way-back-home/set_env.sh

uv run adk deploy cloud_run \
  --project=$GOOGLE_CLOUD_PROJECT \
  --region=$REGION \
  --service_name=mission-analysis-ai \
  --with_ui \
  --a2a \
  ./agent

عندما يُطلب منك إدخال Do you want to continue (Y/n) وAllow unauthenticated invocations to [mission-analysis-ai] (Y/n)?، أدخِل Y في كليهما لنشر وكيل A2A والسماح بالوصول العام إليه.

من المفترَض أن تظهر لك نتيجة مثل:

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

ضبط متغيرات البيئة على Cloud Run

يحتاج الوكيل الذي تم نشره إلى إذن الوصول إلى متغيرات البيئة. تعديل الخدمة:

👉💻 اضبط متغيرات البيئة المطلوبة:

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"

حفظ عنوان URL الخاص بالوكيل

👉💻 الحصول على عنوان URL المنشور:

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\"" >> $HOME/way-back-home/set_env.sh

التحقّق من عملية النشر

👉💻 اختبِر الوكيل الذي تم نشره عن طريق فتح عنوان URL في المتصفّح (تم نشر علامة --with_ui لواجهة الويب الخاصة بـ ADK)، أو اختبِره باستخدام curl:

curl -X GET "$AGENT_URL/list-apps"

من المفترض أن يظهر ردّ يتضمّن وكيلك.

10. الخاتمة

🎉 اكتمل المستوى 1!

يبث جهاز تحديد الموقع الجغرافي الآن بأقصى قوته. يخترق الإشارة المثلثة التشويش الجوي، وهو نبض ثابت يقول "أنا هنا. نجوتُ. تعالَ وابحث عني".

لكنّك لست الشخص الوحيد على هذا الكوكب. عندما ينشط جهاز الإرسال، ستلاحظ أضواء أخرى تومض في الأفق، ما يشير إلى وجود ناجين آخرين ومواقع تحطّم أخرى ومستكشفين آخرين نجوا من الحادث.

في المستوى 2، ستتعلّم كيفية معالجة إشارات SOS الواردة والتنسيق مع الناجين الآخرين. لا يقتصر الإنقاذ على العثور على الشخص، بل يشمل أيضًا العثور على بعضكما البعض.

تحديد المشاكل وحلّها

"لم يتم ضبط MCP_SERVER_URL"

export MCP_SERVER_URL=$(gcloud run services describe location-analyzer \
  --region=$REGION --format='value(status.url)')

لم يتم ضبط PARTICIPANT_ID

source $HOME/way-back-home/set_env.sh
echo $PARTICIPANT_ID

"لم يتم العثور على جدول BigQuery"

uv run python setup/setup_star_catalog.py

"المتخصّصون يطلبون عناوين URL": يعني ذلك أنّ عملية إنشاء النماذج {key} لا تعمل. تحديد:

• هل تم ضبط before_agent_callback على الوكيل الأساسي؟
• هل يتم ضبط قيم حالة إعدادات معاودة الاتصال بشكل صحيح؟
• هل تستخدم الوكلاء الفرعيون {soil_url} (وليس السلاسل f)؟

"تتعارض التحليلات الثلاث" إعادة إنشاء المستند: uv run python generate_evidence.py

"الوكيل لا يستجيب في adk web"

• التأكّد من أنّ المنفذ 8000 صحيح
• التأكّد من ضبط MCP_SERVER_URL وPARTICIPANT_ID
• التحقّق من رسائل الخطأ في نافذة الوحدة الطرفية

ملخّص حول الهندسة المعمارية

المكوّن	النوع	النمط	الغرض
setup_participant_context	معاودة الاتصال	before_agent_callback	استرجاع الإعدادات وضبط الحالة
GeologicalAnalyst	الوكيل	إنشاء نماذج {soil_url}	تصنيف التربة
BotanicalAnalyst	الوكيل	إنشاء نماذج {flora_url}	تصنيف النباتات
AstronomicalAnalyst	الوكيل	{stars_url}, {project_id}	تحديد الموقع الجغرافي باستخدام النجوم
confirm_location	الأداة	الوصول إلى حالة ToolContext	تفعيل جهاز الإرسال
EvidenceAnalysisCrew	ParallelAgent	تركيب الوكيل الفرعي	تشغيل المتخصصين بشكل متزامن
MissionAnalysisAI	الوكيل (الجذر)	المنسّق + معاودة الاتصال	التنسيق والدمج
location-analyzer	خادم FastMCP	MCP مخصّص	التحليل الجيولوجي والنباتي
bigquery.googleapis.com/mcp	OneMCP	Managed MCP	إذن الوصول إلى BigQuery

المفاهيم الرئيسية التي تم إتقانها

‫✓ before_agent_callback: استرجاع الإعدادات قبل تشغيل الوكيل
✓ {key} State Templating: الوصول إلى قيم الحالة في تعليمات الوكيل
✓ ToolContext: الوصول إلى قيم الحالة في وظائف الأدوات
✓ State Sharing: إتاحة الحالة الرئيسية تلقائيًا للوكلاء الفرعيين من خلال InvocationContext
✓ Multi-Agent Architecture: وكلاء متخصصون بمسؤوليات فردية
✓ ParallelAgent: تنفيذ المهام المستقلة بشكل متزامن
✓ Custom MCP Server: خادم MCP خاص بك على Cloud Run
✓ OneMCP BigQuery: نمط MCP مُدار للوصول إلى قاعدة البيانات
✓ Cloud Deployment: نشر بدون حالة باستخدام متغيرات البيئة
✓ A2A Preparation: وكيل جاهز للتواصل بين الوكلاء

لغير اللاعبين: تطبيقات واقعية

تمثّل ميزة "تحديد موقعك الجغرافي بدقة" التحليل المتوازي من الخبراء مع الإجماع، أي تشغيل تحليلات متعدّدة ومتخصّصة باستخدام الذكاء الاصطناعي في الوقت نفسه وتجميع النتائج.

تطبيقات المؤسسات

حالة الاستخدام	Parallel Experts	قاعدة التجميع
التشخيص الطبي	محلّل الصور ومحلّل الأعراض ومحلّل المختبر	الحد الأدنى للدقة 2 من 3
رصد الاحتيال	محلّل المعاملات ومحلّل السلوك ومحلّل الشبكات	أي علامة واحدة = مراجعة
معالجة المستندات	عامل التعرّف الضوئي على الحروف، وعامل التصنيف، وعامل الاستخراج	يجب أن يوافق الجميع
مراقبة الجودة	أداة الفحص المرئي، وأداة تحليل المستشعرات، وأداة التحقّق من المواصفات	بطاقة 2 من 3

الإحصاءات الرئيسية حول تصميم التطبيق

1. before_agent_callback للإعداد: استرجاع الإعداد مرة واحدة في البداية، وتعبئة الحالة لجميع الوكلاء الفرعيين لا تتم قراءة ملف الإعداد في الوكلاء الفرعيين.
2. {key} إنشاء نماذج للحالات: إعلاني، واضح، اصطلاحي لا تستخدم السلاسل المنسّقة أو عمليات الاستيراد أو تعديل sys.path.
3. آليات التوافق: تعالج اتفاقية 2 من 3 حالات الغموض بشكل فعّال بدون الحاجة إلى اتفاق بالإجماع.
4. ParallelAgent للمهام المستقلة: عندما لا تعتمد التحليلات على بعضها البعض، يمكنك تنفيذها في الوقت نفسه لزيادة السرعة.
5. نمطان لمنصّة إدارة الموافقة: مخصّص (يمكنك إنشاء منصّتك الخاصة) مقابل OneMCP (تستضيفها Google). يستخدم كلاهما StreamableHTTP.
6. النشر بدون حالة: يعمل الرمز نفسه محليًا وعند نشره. متغيرات البيئة + واجهة برمجة تطبيقات الخلفية = عدم توفّر ملفات إعداد في الحاويات

الخطوات التالية:

المستوى 2: معالجة إشارة SOS →

تعرَّف على كيفية معالجة إشارات الاستغاثة الواردة من الناجين الآخرين باستخدام أنماط مستندة إلى الأحداث وتنسيق أكثر تقدّمًا بين العملاء.