إنشاء تطبيق ذكاء اصطناعي تفاعلي يمكن لعملائك التفاعل معه من خلال Telegram

1- مقدمة

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

ما ستنشئه

سننشئ عملية تكامل بين تطبيق "مساعد المطعم" المتكامل المستنِد إلى حزمة تطوير الوكلاء (ADK) والمستنِد إلى Gemini، والذي يساعد روّاد المطعم في تصفُّح قائمة المطعم وحجز الطاولات، وتطبيق Telegram للمحادثة. يمكنك التفاعل مع روبوت Telegram وطلب أوصاف باللغة الطبيعية، مثل "أريد طبقًا حارًا ونباتيًا". سيتصل الروبوت بعد ذلك بوكيل ADK الذي يقرأ من قاعدة بيانات Cloud SQL PostgreSQL ويكتب فيها بالكامل من خلال MCP Toolbox for Databases، الذي يتعامل مع جميع عمليات الوصول إلى قاعدة البيانات، بما في ذلك إنشاء عمليات التضمين التلقائية للبحث المتّجه. في هذه الأثناء، سيتمكّن المستخدم من رؤية أنّ الروبوت يقرّ بالرسالة ويكتب ... typing للردّ أثناء انتظار الردّ من وكيل ADK.

ما ستتعلمه

• نشر تطبيق "مساعد المطعم" المتكامل المستنِد إلى حزمة تطوير الوكلاء (ADK) والمستنِد إلى Gemini
• إعداد روبوت دردشة Telegram باستخدام BotFather
• كتابة تطبيقات Python للاستماع إلى خطاف الويب الخاص بالروبوت
• إرسال إجراء المحادثة لعرض الإشعار ... typing في Telegram عند تلقّي رسالة المستخدم، وإجراء عملية الاقتراع لإرسال ... typing بشكلٍ دوري أثناء انتظار الردّ الفعلي
• استدعاء نقطة نهاية Cloud Run الخاصة بـ Restaurant Concierge لمعالجة استفسار المستخدم
• التعامل مع الردّ من وكيل ADK وإرسال رسالة إلى Telegram وإغلاق المخزن المؤقت
• نشر تطبيق Python على Cloud Run
• التفاعل مع روبوت Telegram

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

• حساب على Google Cloud مع حساب فوترة تجريبي
• معرفة أساسية بلغة Python
• ستكون الخبرة السابقة في استخدام حزمة تطوير الوكلاء (ADK) وعمليات النشر على Cloud Run مفيدة
• حساب على Telegram
• (مستحسَن) إكمال الدروس التطبيقية حول الترميز التالية :
• التوليد المعزّز بالاسترجاع (RAG) المستنِد إلى الذكاء الاصطناعي الوكيل باستخدام ADK وMCP Toolbox وCloud SQL: يمكنك مواصلة إنشاء وكيلك من هذا الدرس التطبيقي حول الترميز، والرمز الأولي المقدَّم مطابق
• (أو) الوكلاء على نطاق واسع: بنية متعددة الوكلاء باستخدام بروتوكول A2A على Agent Runtime وعملية التكامل مع حزمة تطوير الوكلاء (ADK): إذا أردت تحسين الوكيل بشكلٍ أكبر باستخدام بنية متعددة الوكلاء

2. إعداد البيئة: مواصلة العمل من الدرس التطبيقي حول الترميز السابق

إنّ السرد الذي نوفّره في هذا الدرس التطبيقي حول الترميز هو في الواقع مواصلة من هذا الدرس التطبيقي حول الترميز الذي يمثّل متطلبًا أساسيًا: التوليد المعزّز بالاسترجاع (RAG) المستنِد إلى الذكاء الاصطناعي الوكيل باستخدام ADK وMCP Toolbox وCloud SQL أو الوكلاء على نطاق واسع: بنية متعددة الوكلاء باستخدام بروتوكول A2A على Agent Runtime وعملية التكامل مع حزمة تطوير الوكلاء (ADK). يمكنك مواصلة عملك من الدرس التطبيقي حول الترميز السابق.

ملاحظة مهمة: إذا لم تكن متأكدًا مما إذا كنت قد أكملت الدرس التطبيقي حول الترميز السابق، يمكنك تخطّي هذا القسم والانتقال إلى Environment Setup - Fresh start with the starter repo

يمكننا البدء في الإنشاء في دليل العمل الخاص بالدرس التطبيقي حول الترميز السابق ( يجب أن يكون دليل العمل build-agent-adk-toolbox-cloudsql أو adk-a2a-agent-runtime-starter). لتجنُّب حدوث أي التباس، لنغيّر اسم الدليل إلى الاسم نفسه الذي نستخدمه عند البدء من جديد.

إذا كنت تواصل العمل من التمرين العملي التوليد المعزّز بالاسترجاع (RAG) المستنِد إلى الذكاء الاصطناعي الوكيل باستخدام ADK وMCP Toolbox وCloud SQL :

mv ~/build-agent-adk-toolbox-cloudsql ~/build-agent-adk-telegram

إذا كنت تواصل العمل من التمرين العملي الوكلاء على نطاق واسع: بنية متعددة الوكلاء باستخدام بروتوكول A2A على Agent Runtime وعملية التكامل مع حزمة تطوير الوكلاء (ADK)

mv ~/adk-a2a-agent-runtime-starter ~/build-agent-adk-telegram

بعد ذلك، لنغيّر دليل العمل إلى هذا الدليل.

cloudshell workspace ~/build-agent-adk-telegram && cd ~/build-agent-adk-telegram
source .env

ملاحظة: إذا تمت إعادة ضبط جلسة Cloud Shell، يمكنك تشغيل bash setup_verify_trial_project.sh && source .env لاستعادة إعدادات مشروعك.

بعد ذلك، تأكَّد من أنّ restaurant-agent قد تم نشره ولديه عنوان URL علني يمكن الوصول إليه.

AGENT_URL=$(gcloud run services describe restaurant-agent \
    --region="$REGION" \
    --format='value(status.url)')

echo "      ✓ Agent service deployed"
echo "      Agent URL: $AGENT_URL"
echo ""

إذا كان بإمكانك الوصول إلى عنوان URL، يمكنك الانتقال إلى القسم التالي: Create Telegram Bot

ملاحظة مهمة: إذا ظهرت لك أخطاء من أي من عمليات التحقّق هذه، ننصحك باتّباع Environment Setup - Fresh start with the starter repo في القسم التالي واستخدام مستودع التعليمات البرمجية الأولية لإجراء عملية إعداد نظيفة.

3. إعداد البيئة: بداية جديدة باستخدام مستودع التعليمات البرمجية الأولية

ملاحظات مهمة:

يتطلب هذا البرنامج التعليمي مشروعًا على Google Cloud مع حساب فوترة نشط. إذا لم يكن لديك حساب بعد، استخدِم البانر في أعلى هذا الدرس التطبيقي حول الترميز للحصول على حساب فوترة تجريبي يمكن استخدامه في هذا البرنامج التعليمي.

ملاحظة مهمة: يمكنك تخطّي هذا القسم إذا كنت تتبع القسم Environment Setup - Continuing from the previous codelab، والانتقال إلى القسم التالي Create Telegram Bot

تُعدّ هذه الخطوة بيئة Cloud Shell وتضبط مشروعك على السحابة الإلكترونية وتستنسخ مستودع التعليمات البرمجية الأولية.

فتح Cloud Shell

افتح Cloud Shell في متصفّحك. توفّر Cloud Shell بيئة تم إعدادها مسبقًا وتتضمّن جميع الأدوات التي تحتاج إليها في هذا الدرس التطبيقي حول الترميز. انقر على تفويض عندما يُطلب منك ذلك.

افتح Cloud Shell

بعد ذلك، انقر على "عرض" -> "محطة الدفع" لفتح محطة الدفع.يجب أن تبدو واجهتك مشابهة لما يلي:

ستكون هذه واجهتنا الرئيسية، مع بيئة التطوير المتكاملة في الأعلى ومحطة الدفع في الأسفل.

إعداد دليل العمل

استنسِخ مستودع التعليمات البرمجية الأولية، حيث يتم تخزين كل الرموز البرمجية التي تكتبها في هذا الدرس التطبيقي حول الترميز:

rm -rf ~/build-agent-adk-telegram
git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git build-agent-adk-telegram
cloudshell workspace ~/build-agent-adk-telegram && cd ~/build-agent-adk-telegram

أنشئ ملف .env من النموذج المقدَّم:

cp .env.example .env

ملاحظة مهمة! تم توفير الخطوة أدناه للمساعدة في إعداد مشروع Google Cloud بسرعة، والذي تم ربطه بحساب فوترة تجريبي (إلزامي). إذا كنت تفضّل استخدام مشروعك السابق، يمكنك تخطّي الخطوة التالية وتنفيذ الخطوة التالية بدلاً منها:

• أضِف اسم مشروعك كمتغيّر GOOGLE_CLOUD_PROJECT في ملف .env.
• فعِّل مشروعك في محطة الدفع باستخدام gcloud config set project your-project-id.

بعد ذلك، يمكنك الانتقال مباشرةً إلى قسم إعداد البنية الأساسية الأولية.

إذا لم تكن متأكدًا من ذلك، يمكنك مواصلة القسم أدناه.

لتبسيط إعداد المشروع في محطة الأوامر، نزِّل نص التهيئة البرمجي هذا في دليل العمل:

curl -sL https://raw.githubusercontent.com/alphinside/cloud-trial-project-setup/main/setup_verify_trial_project.sh -o setup_verify_trial_project.sh

شغِّل النص البرمجي. يتحقّق النص البرمجي من حساب الفوترة التجريبي، وينشئ مشروعًا جديدًا (أو يتحقّق من مشروع حالي)، ويحفظ رقم تعريف مشروعك في ملف .env في الدليل الحالي، ويضبط المشروع النشط في gcloud.

bash setup_verify_trial_project.sh && source .env

سيتولى النص البرمجي تنفيذ ما يلي:

1. التحقّق من أنّ لديك حساب فوترة تجريبي نشط
2. التحقّق من وجود مشروع حالي في .env (إن وُجد)
3. إنشاء مشروع جديد أو إعادة استخدام المشروع الحالي
4. ربط حساب الفوترة التجريبي بمشروعك
5. حفظ رقم تعريف المشروع في .env
6. ضبط المشروع كمشروع gcloud نشط

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

تأكَّد من ضبط المشروع بشكلٍ صحيح من خلال التحقّق من النص الأصفر بجانب دليل العمل في طلب Cloud Shell في محطة الدفع. يجب أن يعرض رقم تعريف مشروعك.

ملاحظة: إذا تمت إعادة ضبط جلسة Cloud Shell في أي وقت أثناء هذا الدرس التطبيقي حول الترميز، انتقِل مرة أخرى إلى دليل العمل وأعِد تشغيل bash setup_verify_trial_project.sh && source .env لاستعادة إعدادات مشروعك. تأكَّد من ظهور نص رقم تعريف المشروع الأصفر مرة أخرى في طلب محطة الدفع.

إعداد البنية الأساسية الأولية

أولاً، علينا تثبيت تبعيات Python باستخدام uv، وهي أداة سريعة لإدارة حِزم Python والمشاريع مكتوبة بلغة Rust ( مستندات uv ). يستخدم هذا الدرس التطبيقي حول الترميز هذه الأداة لضمان السرعة والسهولة في صيانة مشروع Python.

uv sync

بعد ذلك، شغِّل نص التهيئة البرمجي الكامل، الذي ينشئ مثيل Cloud SQL، ويملأ البيانات، وينشر خدمة Toolbox التي ستعمل كحالة أولية لوكيل المطعم.

bash scripts/full_setup.sh > logs/full_setup.log 2>&1 &

ملاحظة: يستغرِق هذا النص البرمجي حوالي 10 إلى 15 دقيقة. يوفّر النص البرمجي مثيل Cloud SQL، وينشئ قاعدة البيانات، ويملأ بيانات القائمة بعمليات التضمين المتّجهة، وينزِّل ويشغِّل ملف Toolbox الثنائي للتطوير المحلي، وينشر وكيل المطعم الأولي وخدمة Toolbox على Cloud Run لتنفيذ عملية نشر متكاملة. سيتم تشغيل النص البرمجي في الخلفية ويمكنك الاطّلاع على السجلّ في الملف logs/full_setup.log.

سيؤدي هذا الإجراء إلى:

• إنشاء مثيل Cloud SQL وملء قاعدة البيانات (المرحلة 1)
• إنشاء إعدادات بيئة الوكيل وبدء خدمة Toolbox المحلية (المرحلة 2)
• نشر خدمات Toolbox والوكيل على Cloud Run (المرحلة 3)

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

source .env
AGENT_URL=$(gcloud run services describe restaurant-agent \
    --region="$REGION" \
    --format='value(status.url)')

echo "      ✓ Agent service deployed"
echo "      Agent URL: $AGENT_URL"
echo ""

افتح واجهة مستخدم مطوّر حزمة تطوير الوكلاء (ADK)، واختَر restaurant_agent، واختبِر باستخدام طلبات بحث مثل المثال التالي:

What Italian dishes do you have?

أو

I want something spicy and creamy

الآن، يتمثل الإجراء التالي في كيفية الانتقال من واجهة تطوير الويب فقط إلى قناة المراسلة على Telegram.

4- إنشاء روبوت Telegram

Telegram هي منصة مراسلة مجانية معروفة تُستخدم على نطاق واسع للتفاعل المستنِد إلى المجتمع. أحد الأسباب هو أنّها توفّر العديد من الطرق للتكامل بسهولة، وبالتالي يمكن للمستخدمين إنشاء روبوتاتهم الخاصة بسهولة باستخدام مجموعة متنوعة من الوظائف المختلفة.

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

استخدام BotFather لإنشاء الروبوت الخاص بك

افتح متصفّح الويب وانتقِل إلى https://telegram.me/BotFather لبدء إنشاء روبوت Telegram الخاص بك.

بدء التفاعل مع BotFather

إرسال الأمر /start

لبدء استخدام BotFather وبدء إنشاء الروبوت الأول، عليك استدعاء الرسالة /start إلى BotFather، وسيشارك بعد ذلك جميع الأوامر التي يمكنك التفاعل معها.

/start

بدء إنشاء الروبوت باستخدام الأمر /newbot

لننشئ الروبوت الجديد من خلال إرسال الأمر /newbot إلى BotFather. سيُطلب منك تسمية الروبوت، ثم سيُطلب منك منح الروبوت username يجب أن ينتهي دائمًا بـ bot . على سبيل المثال، TetrisBot أو tetris_bot. يجب أن يكون اسم المستخدم فريدًا.

بعد إنشاء الروبوت بنجاح، ستتلقّى الرسالة التالية من BotFather.

Done! Congratulations on your new bot. You will find it at t.me/AdkTelegramTest_bot. You can now add a description, about section and profile picture for your bot, see /help for a list of commands. By the way, when you've finished creating your cool bot, ping our Bot Support if you want a better username for it. Just make sure the bot is fully operational before you do this.

Use this token to access the HTTP API:

<YOUR_TELEGRAM_API_KEY>

Keep your token secure and store it safely, it can be used by anyone to control your bot.

For a description of the Bot API, see this page: https://core.telegram.org/bots/api

دوِّن YOUR_TELEGRAM_API_KEY لأنّنا سنستخدمه في القسم التالي.

5- تطوير تطبيق خطاف الويب على Telegram

لنُعدّ دليل العمل لبدء تطوير تطبيق خطاف الويب على Telegram.

mkdir ~/build-agent-adk-telegram/telegram-integration
cd ~/build-agent-adk-telegram

إضافة التبعيات المطلوبة

أنشئ النص البرمجي requirements.txt بالاستناد إلى المحتوى التالي لتوفير تبعيات كافية للنص البرمجي لمستمع خطاف الويب على Telegram.

cloudshell edit ./telegram-integration/requirements.txt

بعد ذلك، أضِف التبعيات التالية:

python-telegram-bot[webhooks]
httpx

إنشاء نص برمجي لمستمع خطاف الويب على Telegram

بعد تثبيت التبعية، يمكننا الآن إنشاء نص برمجي بلغة Python باسم main.py لتطبيق التكامل.

cloudshell edit ~/build-agent-adk-telegram/telegram-integration/main.py

بعد ذلك، انسخ الرمز التالي وألصِقه فيه:

# ./telegram-integration/main.py

import asyncio
import os
import sys
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext
from telegram.constants import ChatAction
import httpx

# Read token from environment variable
TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
ADK_SERVER_URL = os.environ.get("ADK_SERVER_URL", "http://localhost:8000")
ADK_APP_NAME = os.environ.get("ADK_APP_NAME", "restaurant_agent")

# Parse base URL out of ADK_SERVER_URL
BASE_URL = ADK_SERVER_URL.rstrip('/')
if BASE_URL.endswith('/run'):
    BASE_URL = BASE_URL[:-4]
elif BASE_URL.endswith('/query'):
    BASE_URL = BASE_URL[:-6]

if not TOKEN:
    print("Error: TELEGRAM_BOT_TOKEN environment variable not set.")
    print("Please set it before running the application.")
    sys.exit(1)

async def start(update: Update, context: CallbackContext) -> None:
    """Send a message when the command /start is issued."""
    await update.message.reply_text('Hi! I am your ADK Integration Bot. Send me a message and I will forward it to the ADK server.')

async def send_typing_loop(chat_id: int, bot, stop_event: asyncio.Event):
    """Send typing action periodically until the stop event is set."""
    while not stop_event.is_set():
        try:
            await bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)
            # The research suggested repeating every 4 seconds
            await asyncio.sleep(4)
        except Exception as e:
            print(f"Error sending chat action: {e}")
            await asyncio.sleep(1) # Wait a bit before retrying if error

async def handle_message(update: Update, context: CallbackContext) -> None:
    """Handle incoming user messages."""
    user_message = update.message.text
    chat_id = update.message.chat_id
    raw_user_id = str(update.message.from_user.id)
    
    # Derive unique user_id and session_id for this user
    user_id = f"tg_{raw_user_id}"
    session_id = f"tg_sess_{raw_user_id}"

    print(f"Received message from {user_id}: {user_message}")

    # Create a stop event for the typing loop
    stop_event = asyncio.Event()
    
    # Start the typing loop as a background task
    typing_task = asyncio.create_task(send_typing_loop(chat_id, context.bot, stop_event))

    try:
        async with httpx.AsyncClient() as client:
            # 1. Check if the session exists
            session_url = f"{BASE_URL}/apps/{ADK_APP_NAME}/users/{user_id}/sessions/{session_id}"
            session_check = await client.get(session_url, timeout=10.0)
            
            if session_check.status_code == 404:
                # 2. If session doesn't exist, create it
                print(f"Session {session_id} not found. Creating session...")
                session_create = await client.post(session_url, json={}, timeout=10.0)
                if session_create.status_code != 200:
                    raise Exception(f"Failed to create session: {session_create.status_code} {session_create.text}")
            elif session_check.status_code != 200:
                raise Exception(f"Error checking session: {session_check.status_code} {session_check.text}")
            
            # 3. Run the ADK agent
            run_url = f"{BASE_URL}/run"
            payload = {
                "appName": ADK_APP_NAME,
                "userId": user_id,
                "sessionId": session_id,
                "newMessage": {
                    "role": "user",
                    "parts": [{"text": user_message}]
                }
            }
            response = await client.post(run_url, json=payload, timeout=60.0)
            
        if response.status_code == 200:
            events = response.json()
            if isinstance(events, list) and len(events) > 0:
                # The last event contains the final text response
                last_event = events[-1]
                content = last_event.get("content", {})
                parts = content.get("parts", [])
                if parts and "text" in parts[0]:
                    reply_text = parts[0]["text"]
                else:
                    reply_text = "ADK agent returned an empty or non-text response."
            else:
                reply_text = "No events returned from ADK agent."
        else:
            reply_text = f"Error communicating with ADK server (Status: {response.status_code})."
            
    except Exception as e:
        reply_text = f"Failed to connect to ADK server: {e}"
    finally:
        # Stop the typing loop
        stop_event.set()
        await typing_task

    # Send the final response back to the user
    await update.message.reply_text(reply_text)

def main() -> None:
    """Start the bot."""
    # Create the Application and pass it your bot's token.
    application = Application.builder().token(TOKEN).build()

    # on different commands - answer in Telegram
    application.add_handler(CommandHandler("start", start))

    # on non command i.e message - echo the message on Telegram
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    # Check if running in webhook mode (e.g., on Cloud Run)
    port = os.environ.get("PORT")
    service_url = os.environ.get("SERVICE_URL")

    if port and service_url:
        if not service_url.startswith("http"):
            service_url = f"https://{service_url}"
        
        print(f"Starting bot in WEBHOOK mode on port {port} with url {service_url}")
        
        application.run_webhook(
            listen="0.0.0.0",
            port=int(port),
            url_path=TOKEN,
            webhook_url=f"{service_url}/{TOKEN}",
            allowed_updates=Update.ALL_TYPES
        )
    else:
        print("Starting bot in POLLING mode")
        # Run the bot until the user presses Ctrl-C
        application.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()

فهم الرمز البرمجي لعملية التكامل مع روبوت Telegram

عندما يرسل المستخدم رسالة، يتم تشغيل البنية الأساسية لبرنامج معالجة البيانات التالية ضمن handle_message():

الخطوة 1: استنتاج الهوية والجلسة

يربط الروبوت رقم تعريف مستخدم Telegram بمعرّفات فريدة لحزمة تطوير الوكلاء (ADK) للحفاظ على تمييز جلسات المستخدمين:

user_id = f"tg_{raw_user_id}"
session_id = f"tg_sess_{raw_user_id}"

الخطوة 2: حالة "جارٍ الكتابة" غير المتزامنة (الأسطر 53–58)

لضمان تجربة مستخدم سريعة الاستجابة أثناء معالجة وكيل ADK للطلب (الذي قد يستغرِق عدة ثوانٍ)، يبدأ الروبوت حلقة غير متزامنة في الخلفية:

• يتم إنشاء asyncio.Event باسم stop_event.
• تنشئ asyncio.create_task send_typing_loop(...) في الخلفية.
• تُرسِل الحلقة إجراء ChatAction.TYPING إلى Telegram كل 4 ثوانٍ إلى أن يتم ضبط stop_event.

الخطوة 3: التحقّق من جلسة ADK وإنشاؤها (الأسطر 61–72)

قبل تنفيذ الوكيل، يتحقّق الروبوت مما إذا كانت هناك جلسة حالية:

1. يُرسِل طلب GET إلى /apps/{appName}/users/{userId}/sessions/{sessionId}.
2. إذا كان الردّ 404 Not Found، يتم إنشاء الجلسة من خلال طلب POST إلى عنوان URL نفسه مع نص JSON فارغ.
3. إذا تم عرض حالة أخرى غير 200 أو 404، يتم طرح استثناء.

الخطوة 4: إرسال الطلب إلى الوكيل (الأسطر 74–85)

تتم إعادة توجيه حمولة الرسالة إلى نقطة النهاية /run في حزمة تطوير الوكلاء (ADK):

• نقطة النهاية: POST /run
• تم ضبط مهلة الطلب على 60.0 ثانية للسماح بالاستدلال المعقّد أو وقت الاستجابة في المصدر.
• بنية الحمولة:

{
  "appName": "restaurant_agent",
  "userId": "tg_<user_id>",
  "sessionId": "tg_sess_<user_id>",
  "newMessage": {
    "role": "user",
    "parts": [{"text": "<user_message>"}]
  }
}

الخطوة 5: تحليل الردّ (الأسطر 87–101)

يعرض خادم حزمة تطوير الوكلاء (ADK) قائمة بأحداث الرسائل. يفحص الروبوت الصفيف المعروض:

• يستردّ الروبوت الحدث الأخير في القائمة (events[-1]).
• ينتقِل الروبوت إلى المحتوى النصي من خلال event["content"]["parts"][0]["text"].
• إذا لم يتم عرض أي أحداث أو إذا كانت بنية النص غير متوفّرة، يتم ضبط نص عنصر نائب وصفي.

الخطوة 6: الإيقاف المؤقت وإرسال الردّ (الأسطر 103–111)

• في القسم finally، يتم ضبط stop_event، ما يؤدي إلى إيقاف حلقة إجراء الكتابة.
• ينتظر الروبوت اكتمال typing_task لضمان تنظيف الموارد.
• أخيرًا، يردّ الروبوت على محادثة Telegram باستخدام نص الردّ الذي تم تحليله.

6- نشر تطبيق خطاف الويب على Telegram على Cloud Run

بعد ذلك، سننشر مستمع خطاف الويب على Telegram على Cloud Run، حتى يتمكّن الروبوت من التواصل معه.

إنشاء ملف Dockerfile

أولاً، علينا إنشاء ملف Dockerfile.

cloudshell edit ~/build-agent-adk-telegram/telegram-integration/Dockerfile

بعد ذلك، انسخ الرمز التالي وألصِقه فيه:

# Use an official Python runtime as a parent image
FROM python:3.11-slim

# Prevent Python from writing pyc files to disc and buffering stdout/stderr
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Set the working directory in the container
WORKDIR /app

# Install system dependencies if needed
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Copy the dependencies file to the working directory
COPY requirements.txt .

# Install any needed packages specified in requirements.txt
RUN pip install --no-cache-dir -r requirements.txt

# Copy the rest of the application code
COPY main.py .

# Expose the port that Cloud Run will provide via environment variable
EXPOSE 8080

# Run main.py when the container launches
CMD ["python", "main.py"]

يتم وضع الخدمة في حاوية باستخدام python:3.11-slim للحفاظ على صغر حجم الصورة:

• يتم تثبيت التبعيات من requirements.txt (python-telegram-bot[webhooks] وhttpx).
• يتم عرض المنفذ العادي 8080.
• يتم تشغيل python main.py.

إعداد متغيرات البيئة

بعد ذلك، لنُعدّ التحقّق مما إذا تم نشر وكيلنا بنجاح.

AGENT_URL=$(gcloud run services describe restaurant-agent \
    --region="$REGION" \
    --format='value(status.url)')

echo "      ✓ Agent service deployed"
echo "      Agent URL: $AGENT_URL"
echo ""

بعد ذلك، لنضع TELEGRAM_BOT_TOKEN الذي حصلنا عليه سابقًا في .env.

ملاحظات: لا تنسَ استبدال YOUR_TELEGRAM_API_KEY بمفتاحك الفعلي.

echo "TELEGRAM_BOT_TOKEN=YOUR_TELEGRAM_API_KEY" >> .env

بعد ذلك، لنملأ بيانات .env بالقيم الأخرى التي نحتاج إليها.

echo "ADK_SERVER_URL=$AGENT_URL" >> .env
echo "ADK_APP_NAME=restaurant_agent" >> .env
echo "SERVICE_NAME=telegram-integration" >> .env
source .env

إنشاء نص برمجي للنشر

لننشئ نصًا برمجيًا للنشر يوفّر عمليات تحقّق كاملة وينشر التطبيق على Cloud Run.

cloudshell edit ~/build-agent-adk-telegram/telegram-integration/deploy.sh

انسخ الرمز التالي وألصِقه في الملف:

#!/usr/bin/env bash
# ./telegram-integration/deploy.sh

# Exit immediately if a command exits with a non-zero status
set -euo pipefail

# Color codes for neat terminal output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0;37m' # No Color
# Load environment variables from .env if it exists
if [ -f .env ]; then
    echo -e "${GREEN}✔ Loading environment variables from .env...${NC}"
    export $(grep -v '^#' .env | xargs)
fi

echo -e "${BLUE}====================================================${NC}"
echo -e "${BLUE}   Google Cloud Run Deployment: Telegram Bot        ${NC}"
echo -e "${BLUE}====================================================${NC}"

# 1. Check for gcloud CLI
if ! command -v gcloud &> /dev/null; then
    echo -e "${RED}Error: 'gcloud' CLI is not installed.${NC}"
    echo "Please install the Google Cloud SDK and try again."
    echo "See: https://cloud.google.com/sdk/docs/install"
    exit 1
fi

# 2. Check active gcloud account/auth
ACTIVE_ACCOUNT=$(gcloud auth list --filter=status:ACTIVE --format="value(account)" 2>/dev/null || true)
if [ -z "$ACTIVE_ACCOUNT" ]; then
    echo -e "${RED}Error: No active Google Cloud account found.${NC}"
    echo "Please run: gcloud auth login"
    exit 1
fi

# 3. Detect / Prompt for GCP Project
DEFAULT_PROJECT=${GCP_PROJECT_ID:-$(gcloud config get-value project 2>/dev/null || true)}
if [ -n "${DEFAULT_PROJECT}" ]; then
    echo -e "${GREEN}✔ Using GCP Project: $DEFAULT_PROJECT${NC}"
    GCP_PROJECT="$DEFAULT_PROJECT"
else
    echo -n "Enter GCP Project ID: "
    read -r GCP_PROJECT
fi

if [ -z "$GCP_PROJECT" ]; then
    echo -e "${RED}Error: GCP Project ID is required.${NC}"
    exit 1
fi

# Set active project
gcloud config set project "$GCP_PROJECT" &> /dev/null

# 4. Configure Service Parameters
DEFAULT_SERVICE=${SERVICE_NAME:-"telegram-integration"}
if [ -n "${SERVICE_NAME:-}" ]; then
    echo -e "${GREEN}✔ Using Cloud Run Service Name: $SERVICE_NAME${NC}"
else
    echo -n "Enter Cloud Run Service Name [Default: $DEFAULT_SERVICE]: "
    read -r SERVICE_NAME
    SERVICE_NAME=${SERVICE_NAME:-$DEFAULT_SERVICE}
fi

DEFAULT_REGION=${REGION:-"us-central1"}
if [ -n "${REGION:-}" ]; then
    echo -e "${GREEN}✔ Using Cloud Run Region: $REGION${NC}"
else
    echo -n "Enter Cloud Run Region [Default: $DEFAULT_REGION]: "
    read -r REGION
    REGION=${REGION:-$DEFAULT_REGION}
fi

DEFAULT_ADK_APP=${ADK_APP_NAME:-"restaurant_agent"}
if [ -n "${ADK_APP_NAME:-}" ]; then
    echo -e "${GREEN}✔ Using ADK App Name: $ADK_APP_NAME${NC}"
    ADK_APP="$ADK_APP_NAME"
else
    echo -n "Enter ADK App Name [Default: $DEFAULT_ADK_APP]: "
    read -r ADK_APP
    ADK_APP=${ADK_APP:-$DEFAULT_ADK_APP}
fi

# 5. Retrieve/Prompt for Telegram Bot Token
if [ -n "${TELEGRAM_BOT_TOKEN:-}" ]; then
    echo -e "${GREEN}✔ Found TELEGRAM_BOT_TOKEN in environment.${NC}"
    BOT_TOKEN="$TELEGRAM_BOT_TOKEN"
else
    echo -e "${YELLOW}TELEGRAM_BOT_TOKEN is not set in your environment.${NC}"
    echo -n "Enter your Telegram Bot Token (input will be hidden): "
    read -s -r BOT_TOKEN
    echo ""
fi

if [ -z "$BOT_TOKEN" ]; then
    echo -e "${RED}Error: Telegram Bot Token is required.${NC}"
    exit 1
fi

# 6. Retrieve/Prompt for ADK Server URL
DEFAULT_ADK_URL="http://localhost:8000"
if [ -n "${ADK_SERVER_URL:-}" ]; then
    echo -e "${GREEN}✔ Found ADK_SERVER_URL in environment: $ADK_SERVER_URL${NC}"
    ADK_URL="$ADK_SERVER_URL"
else
    echo -n "Enter your ADK Server URL [Default: $DEFAULT_ADK_URL]: "
    read -r ADK_URL
    ADK_URL=${ADK_URL:-$DEFAULT_ADK_URL}
fi

# Enable required GCP services
echo -e "\n${YELLOW}Checking and enabling required GCP services...${NC}"
gcloud services enable run.googleapis.com cloudbuild.googleapis.com artifactregistry.googleapis.com --project "$GCP_PROJECT"

# Determine source directory dynamically
SOURCE_DIR="."
if [ -d "telegram-integration" ]; then
    SOURCE_DIR="telegram-integration"
    echo -e "${GREEN}✔ Found source directory: telegram-integration${NC}"
elif [ -f "Dockerfile" ]; then
    SOURCE_DIR="."
    echo -e "${GREEN}✔ Dockerfile found in current directory. Using current directory as source.${NC}"
else
    echo -e "${RED}Error: Could not find source directory 'telegram-integration' or Dockerfile in current directory.${NC}"
    exit 1
fi

# 7. First-pass Deployment with placeholder SERVICE_URL
# This boots the container in Webhook mode (so health check binds to port)
# but uses a high-reliability placeholder URL (google.com) to pass DNS verification checks.
echo -e "\n${YELLOW}Deploying to Cloud Run (Step 1/2: Initial Deploy)...${NC}"
gcloud run deploy "$SERVICE_NAME" \
  --source "$SOURCE_DIR" \
  --region "$REGION" \
  --allow-unauthenticated \
  --set-env-vars "TELEGRAM_BOT_TOKEN=$BOT_TOKEN,ADK_SERVER_URL=$ADK_URL,ADK_APP_NAME=$ADK_APP,SERVICE_URL=https://google.com" \
  --project "$GCP_PROJECT"

# 8. Retrieve the actual service URL
echo -e "\n${YELLOW}Retrieving service URL...${NC}"
SERVICE_URL=$(gcloud run services describe "$SERVICE_NAME" --region "$REGION" --project "$GCP_PROJECT" --format 'value(status.url)')
echo -e "${GREEN}✔ Service URL is: $SERVICE_URL${NC}"

# 9. Update service environment variables with the real SERVICE_URL
# This triggers a rolling update and registers the correct webhook with Telegram automatically!
echo -e "\n${YELLOW}Updating configuration with final Webhook URL (Step 2/2)...${NC}"
gcloud run services update "$SERVICE_NAME" \
  --region "$REGION" \
  --set-env-vars "TELEGRAM_BOT_TOKEN=$BOT_TOKEN,ADK_SERVER_URL=$ADK_URL,ADK_APP_NAME=$ADK_APP,SERVICE_URL=$SERVICE_URL" \
  --project "$GCP_PROJECT"

echo -e "\n${GREEN}====================================================${NC}"
echo -e "${GREEN}   Deployment Completed Successfully! 🎉            ${NC}"
echo -e "${GREEN}====================================================${NC}"
echo -e "Service Name:   ${BLUE}$SERVICE_NAME${NC}"
echo -e "Region:         ${BLUE}$REGION${NC}"
echo -e "Active URL:     ${BLUE}$SERVICE_URL${NC}"
echo -e "Webhook Path:   ${BLUE}$SERVICE_URL/<bot-token>${NC}"
echo -e "ADK Backend:    ${BLUE}$ADK_URL${NC}"
echo -e "ADK App Name:   ${BLUE}$ADK_APP${NC}"
echo -e "${GREEN}====================================================${NC}"
echo "Your Telegram Bot has been configured to use webhooks."
echo "Any message sent to your bot will now trigger this Cloud Run instance."

النص البرمجي للنشر المزدوج (deploy.sh)

عند النشر على Google Cloud Run، يحتاج الروبوت إلى تحديد عنوان URL الخاص به (SERVICE_URL) في بيئته حتى يتمكّن من تسجيله كهدف لخطاف الويب على Telegram. لحلّ هذه التبعية الدائرية (عنوان URL غير معروف إلى أن يتم النشر، ولكنّ الخدمة تتطلّب عنوان URL للتشغيل بدون حدوث أخطاء في عملية التحقّق من الصحة)، ينفّذ deploy.sh عملية نشر على مرحلتَين:

1. الخطوة 1: النشر الأولي: يتم تشغيل الحاوية باستخدام نظام أسماء نطاقات (DNS) عنصر نائب (https://google.com) حتى تبدأ الخدمة بنجاح، وترتبط بالمنفذ المحلي، وتجتاز عمليات التحقّق الأولية من الصحة على Cloud Run.
2. الخطوة 2: استرداد عنوان URL: يتم استخراج نقطة نهاية Cloud Run التي تم إنشاؤها حديثًا بشكلٍ آلي باستخدام gcloud run services describe.
3. الخطوة 3: تعديل الإعدادات: يتم تعديل متغيرات البيئة باستخدام عنوان URL الفعلي للخدمة المباشرة. يؤدي ذلك إلى بدء عملية تعديل متدرّجة نظيفة في Cloud Run وتسجيل هدف خطاف الويب الصحيح بأمان في Telegram API.

النشر على Cloud Run

يطبع النص البرمجي للنشر عنوان URL الخاص بالوكيل. افتح عنوان URL في متصفّحك للوصول إلى واجهة مستخدم مطوّر حزمة تطوير الوكلاء (ADK) نفسها التي يتم تشغيلها على Cloud Run.

cd ~/build-agent-adk-telegram
bash ./telegram-integration/deploy.sh

إذا سارت الأمور على ما يُرام، يمكنك الآن بدء التحدّث إلى الروبوت مباشرةً من تطبيق محادثة Telegram، والعثور على الروبوت الذي أنشأته للتو وبدء التفاعل معه:

What Italian dishes do you have?

أو

I want something spicy and creamy

شاهِد الروبوت وهو يرسل الحالة "...جارٍ الكتابة"، ثم سيعرض قريبًا الرسالة من حزمة تطوير الوكلاء (ADK) التي أنشأتها سابقًا.

7- تهانينا!

لقد أنشأت ونشرت ودمجت بالكامل مساعد قائمة المطعم الذكي المستنِد إلى حزمة تطوير الوكلاء (ADK) والوكيل المستنِد إلى الذكاء الاصطناعي مع Telegram، من خلال عملية التواصل بين خادم عميل HTTP، وسمحت للمستخدمين بالاستعلام عن قائمتهم المفضّلة وحجز المطعم.

ما تعلمته

• نشر تطبيق "مساعد المطعم" والوكيل المستنِد إلى حزمة تطوير الوكلاء (ADK) وMCP Toolbox وإعدادها على Cloud Run
• كيفية إعداد روبوت Telegram باستخدام BotFather
• كيفية كتابة نصوص Python البرمجية للاستماع إلى خطاف الويب على Telegram والتفاعل مع وكيل ADK لتمرير طلبات المستخدمين والردّ عليها وفقًا لذلك
• كيفية تنفيذ "... typing" في Telegram للإشارة إلى أنّه تتم معالجة الرسائل كتعليقات في الوقت الفعلي للمستخدمين أثناء انتظار ردّ وكيل ADK
• كيفية نشر نص Python البرمجي على Cloud Run والقدرة على التفاعل معه

إخلاء مساحة

لتجنُّب تحمّل رسوم في حسابك على Google Cloud، احذف الموارد التي تم إنشاؤها في هذا الدرس التطبيقي حول الترميز.

الخيار 1: حذف المشروع (مستحسَن)

gcloud projects delete $GOOGLE_CLOUD_PROJECT

ملاحظة مهمة: يمكنك التراجع عن عملية الحذف هذه لفترة محدودة من خلال استخدام هذا الأمر gcloud projects undelete ${GOOGLE_CLOUD_PROJECT}

الخيار 2: حذف موارد فردية

# If you follow from previous A2A Agent Runtime codelab
# Delete the Agent Runtime deployment (skip if not found)
uv run python -c "
import vertexai
from google.genai import types
vertexai.init(project='$GOOGLE_CLOUD_PROJECT', location='$REGION')
client = vertexai.Client(
    project='$GOOGLE_CLOUD_PROJECT', location='$REGION',
    http_options=types.HttpOptions(api_version='v1beta1'),
)
try:
    agent = client.agent_engines.get(name='$RESERVATION_AGENT_RESOURCE_NAME')
    agent.delete(force=True)
    print('Agent Runtime deployment deleted.')
except Exception as e:
    print(f'No agent deployment found or already deleted, skipping. ({e})')
"

# Delete GCS staging bucket (skip if STAGING_BUCKET is not set)
if [ -n "$STAGING_BUCKET" ]; then
  gsutil rm -r gs://$STAGING_BUCKET
else
  echo "STAGING_BUCKET not set, skipping bucket deletion."
fi

# Delete Cloud Run services
gcloud run services delete restaurant-agent --region=$REGION --quiet
gcloud run services delete toolbox-service --region=$REGION --quiet
gcloud run services delete telegram-integration --region=$REGION --quiet

# Delete Cloud SQL instance
gcloud sql instances delete $DB_INSTANCE --quiet