Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

お客様が Telegram を介して操作できるエージェント型 AI アプリを構築する

1. はじめに

お客様がすでに使用しているメッセージングアプリケーションから直接利用できる、シームレスでインタラクティブなエージェント型 AI エクスペリエンスを作成します。ウェブインターフェースや最新のメッセージングチャネルでスムーズに動作するインテリジェントアプリケーションを開発してデプロイする方法について説明します。

作成するアプリの概要

Gemini を搭載した ADK ベースのアプリケーション「レストランコンシェルジュ」と Telegram チャットアプリを統合します。レストランコンシェルジュは、ユーザーがレストランのメニューを閲覧して予約するのに役立ちます。Telegram ボットとやり取りして、「スパイシーなベジタリアン料理が食べたい」などの自然言語の説明をリクエストできます。ボットは ADK エージェントに接続します。ADK エージェントは、データベース向け MCP ツールボックスを介して Cloud SQL PostgreSQL データベースから読み取り、書き込みを行います。データベース向け MCP ツールボックスは、ベクトル検索用の自動エンベディング生成など、すべてのデータベースアクセスを処理します。ユーザーは、ADK エージェントからのレスポンスを待っている間、ボットがメッセージを認識して ... typing と入力していることを確認できます。

学習内容

Gemini を搭載した ADK ベースのアプリケーション「レストランコンシェルジュ」をデプロイする
BotFather を使用して Telegram チャットボットを設定する
ボットのウェブフックをリッスンする Python アプリケーションを作成する
ユーザーメッセージで Telegram に ... typing 通知を表示するチャットアクションを送信し、実際のレスポンスを待っている間、定期的に ... typing を送信するポーリングを行う
Restaurant Concierge Cloud Run エンドポイントを呼び出してユーザーの問い合わせを処理する
ADK エージェントからの戻り値を処理し、Telegram にメッセージを送信してバッファを閉じる
Python アプリケーションを Cloud Run にデプロイする
Telegram ボットとやり取りする

前提条件

トライアル請求先アカウントを持つ Google Cloud アカウント
Python に関する基本的な知識
ADK と Cloud Run のデプロイの経験があると役立ちます
Telegram アカウント
（推奨）次の Codelab を完了していること :
ADK、MCP ツールボックス、Cloud SQL を使用したエージェント RAG -> この Codelab からエージェントの構築を続行できます。提供されているスターターコードは同じです。
（または）エージェントのスケール: Agent Runtime と ADK 統合での A2A プロトコルを使用したマルチエージェントアーキテクチャ -> マルチエージェントアーキテクチャを使用してエージェントをさらに強化する場合

2. 環境の設定 - 前の Codelab からの継続

この Codelab で提供する説明は、実際にはこの前提条件の Codelab（ADK、MCP ツールボックス、Cloud SQL を使用したエージェント RAG またはエージェントのスケール: Agent Runtime と ADK 統合での A2A プロトコルを使用したマルチエージェントアーキテクチャ）の続きです。前の Codelab から作業を続行できます。

前の Codelab の作業ディレクトリ（作業ディレクトリは build-agent-adk-toolbox-cloudsql または adk-a2a-agent-runtime-starter）でビルドを開始できます。混乱を避けるため、新規スタート時に使用するディレクトリ名と同じディレクトリ名に変更しましょう。

ラボ ADK、MCP ツールボックス、Cloud SQL を使用したエージェント RAG から続行する場合 :

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

ラボエージェントのスケール: Agent Runtime と ADK 統合での A2A プロトコルを使用したマルチエージェントアーキテクチャから続行する場合

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

次に、作業ディレクトリをそのディレクトリに変更します。

cloudshell workspace ~/build-agent-adk-telegram && cd ~/build-agent-adk-telegram
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 に進みます。

3. 環境の設定 - スターターリポジトリを使用した新規スタート

このステップでは、Cloud Shell 環境を準備し、Google Cloud プロジェクトを構成して、スターターリポジトリをクローンします。

Cloud Shell を開く

ブラウザで Cloud Shell を開きます。Cloud Shell には、この Codelab で必要なすべてのツールがプリインストールされた環境が用意されています。プロンプトが表示されたら、[Authorize] をクリックします。

[View] -> [Terminal] をクリックしてターミナルを開きます。インターフェースは次のようになります。

これがメインのインターフェースになります。上部に IDE、下部にターミナルが表示されます。

作業ディレクトリを設定する

スターターリポジトリをクローンします。この Codelab で作成するすべてのコードはここに保存されます。

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

ターミナルでのプロジェクトの設定を簡素化するには、このプロジェクト設定スクリプトを作業ディレクトリにダウンロードします。

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

スクリプトを実行します。トライアル請求先アカウントを確認し、新しいプロジェクトを作成（または既存のプロジェクトを検証）して、プロジェクト ID を現在のディレクトリの .env ファイルに保存し、gcloud でアクティブなプロジェクトを設定します。

bash setup_verify_trial_project.sh && source .env

このスクリプトによって行われる処理は次のとおりです。

有効なトライアル請求先アカウントがあることを確認する
.env に既存のプロジェクトがあるかどうかを確認する（ある場合）
新しいプロジェクトを作成するか、既存のプロジェクトを再利用する
トライアル請求先アカウントをプロジェクトにリンクする
プロジェクト ID を .env に保存する
プロジェクトをアクティブな gcloud プロジェクトとして設定する

プロジェクトが正しく設定されていることを確認するには、Cloud Shell ターミナルプロンプトの作業ディレクトリの横にある黄色のテキスト を確認します。プロジェクト ID が表示されます。

スターターインフラストラクチャの設定

まず、uv を使用して Python の依存関係をインストールする必要があります。これは、Rust で記述された高速な Python パッケージおよびプロジェクトマネージャーです（uv ドキュメント）。この Codelab では、Python プロジェクトの保守を高速かつ簡単に行うために使用します。

uv sync

次に、完全な設定スクリプトを実行します。これにより、Cloud SQL インスタンスが作成され、データがシードされ、レストランエージェントの初期状態として機能するツールボックスサービスがデプロイされます。

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

これにより、次のことが行われます。

Cloud SQL インスタンスを作成してデータベースをシードする（フェーズ 1）
エージェント環境構成を生成してローカルツールボックスサービスを開始する（フェーズ 2）
ツールボックスサービスとエージェントサービスを Cloud Run にデプロイする（フェーズ 3）

このデプロイが完了したら、Cloud Run URL で ADK 開発 UI にアクセスできます。

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 開発 UI を開き、restaurant_agent を選択して、次の例のようなクエリでテストします。

What Italian dishes do you have?

または

I want something spicy and creamy

次のアクションは、ウェブ開発インターフェースから Telegram メッセージングチャネルに移行する方法です。

4. Telegram ボットを作成する

Telegram は、コミュニティベースのエンゲージメントに広く使用されている無料のメッセージングプラットフォームです。その理由の 1 つは、簡単に統合できる方法が多数用意されているため、さまざまな機能を持つ独自のボットを簡単に作成できることです。

ここでは、BotFather を使用して初めて独自の bot を作成します。このセッションでは Telegram を使用しますが、同じ方法を WhatsApp やその他のメッセージングプラットフォームでも使用できます。

BotFather を使用して独自のボットを作成する

ウェブブラウザを開き、 https://telegram.me/BotFather にアクセスして、独自の Telegram ボットの作成を開始します。

BotFather とのやり取りを開始する

/start コマンドを送信する

BotFather を使用して最初のボットの作成を開始するには、BotFather に /start メッセージを送信する必要があります。これにより、BotFather が提供するすべてのコマンドを共有して、さらにやり取りできるようになります。

/start

/newbot コマンドでボットの作成を開始する

BotFather に /newbot コマンドを送信して、新しいボットを作成しましょう。ボットの名前を入力するように求められます。次に、ボットに username を指定するように求められます。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 Webhook アプリケーションを開発する

作業ディレクトリを準備して、Telegram Webhook アプリケーションの開発を開始しましょう。

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

必要な依存関係を追加する

次の内容で requirements.txt スクリプトを作成し、Telegram Webhook リスナースクリプトに必要な依存関係を提供します。

cloudshell edit ./telegram-integration/requirements.txt

次に、次の依存関係を追加します。

python-telegram-bot[webhooks]
httpx

Telegram Webhook リスナーのスクリプトを作成する

依存関係がインストールされたら、統合アプリケーション用の 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: ID とセッションの派生

ボットは、Telegram ユーザー ID を一意の 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(...) を生成します。
ループは、stop_event が設定されるまで 4 秒ごとに ChatAction.TYPING アクションを Telegram に送信します。

ステップ 3: ADK セッションの検証と作成（61 ～ 72 行）

エージェントを実行する前に、ボットはセッションがすでに存在するかどうかを確認します。

/apps/{appName}/users/{userId}/sessions/{sessionId} に GET リクエストを送信します。
レスポンスが 404 Not Found の場合は、空の JSON 本文を含む同じ URL への POST リクエストでセッションを作成します。
200 または 404 以外のステータスが返された場合は、例外が発生します。

ステップ 4: エージェントにリクエストを送信する（74 ～ 85 行）

メッセージペイロードは ADK /run エンドポイントに転送されます。

エンドポイント: 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 Webhook アプリケーションを Cloud Run にデプロイする

次に、Telegram Webhook リスナーを 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 に配置します。

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 は 2 段階のデプロイを実行します。

ステップ 1: 初期デプロイ: プレースホルダ DNS（https://google.com）でコンテナを起動し、サービスが正常に起動してローカルポートにバインドされ、初期の Cloud Run ヘルスチェックに合格するようにします。
ステップ 2: URL を取得する: gcloud run services describe を使用して、新しく作成された Cloud Run エンドポイントをプログラムで抽出します。
ステップ 3: 構成を更新する: 実際のライブサービス URL で環境変数を更新します。これにより、Cloud Run でクリーンなローリングアップデートがトリガーされ、正しいウェブフックターゲットが Telegram API に安全に登録されます。

Cloud Run へのデプロイ

デプロイスクリプトはエージェント URL を出力します。ブラウザで開いて、Cloud Run で実行されている同じ ADK 開発 UI にアクセスします。

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

すべてがうまくいけば、Telegram チャットアプリケーションからボットに直接チャットを開始できます。作成したボットを見つけて、やり取りを開始します。

What Italian dishes do you have?

または

I want something spicy and creamy

ボットがステータス「...is typing」を送信し、すぐに以前に作成した ADK からメッセージが返されることを確認します。

7. 完了

HTTP クライアントサーバー通信を介して、スマートレストランメニューアシスタント ADK ベースの AI エージェントを構築、デプロイ、Telegram と完全に統合し、ユーザーがお気に入りのメニューを検索してレストランを予約できるようにしました。

学習した内容

レストランコンシェルジュ、ADK ベースのエージェント、MCP ツールボックスを Cloud Run にデプロイして構成する
BotFather を使用して Telegram ボットを設定する方法
Telegram Webhook をリッスンし、ADK エージェントとやり取りしてユーザーのクエリを渡して適切に応答する Python スクリプトを作成する方法
ADK エージェントからのレスポンスを待っている間、メッセージが処理中であることをユーザーにリアルタイムフィードバックとして通知する "... typing" を Telegram に実装する方法。
Python スクリプトを Cloud Run にデプロイして操作できるようにする方法

クリーンアップ

Google Cloud アカウントに課金されないようにするには、この Codelab で作成したリソースを削除します。

オプション 1: プロジェクトを削除する（推奨）

gcloud projects delete $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