Note
.NETの Telegram ボットのセルフホスティング ヘルパーは近日公開予定です。
Note
Telegram ボットのセルフホスティング ヘルパーは、現在 Go では使用できません。
agent-framework-hosting-telegram は、Telegram Bot API の更新を Agent Framework の実行値に変換し、最終的な実行またはストリーミング実行を Bot API 操作としてレンダリングします。 ボット クライアント、ポーリング ランタイム、Webhook ルーター、コマンド レジストリ、または配信フレームワークは提供されません。
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-telegram azure-identity
更新ペイロードを提供し、ヘルパーによって返される操作を実行できる Telegram クライアント ライブラリを使用します。 このサンプルでは aiogramを使用していますが、ヘルパーはそれに関連付けされていません。
更新プログラムを処理する
aiogram webhook サンプルでは、Telegram のシークレット ヘッダーを検証し、更新プログラムをディスパッチし、ボット スコープのセッション ID を使用して、プライベート チャットまたは共有グループ チャットごとにエージェント セッションを保持します。
async def handle_update(update: Mapping[str, Any]) -> None:
"""Process one Telegram update through the sample agent."""
callback_query_id = telegram_callback_query_id(update)
if callback_query_id is not None:
await bot.answer_callback_query(callback_query_id=callback_query_id)
chat_id = telegram_chat_id(update)
session_id = telegram_session_id(update, bot_id=bot.id)
if chat_id is None or session_id is None:
return
# Background webhook tasks may overlap. Serialize each chat so /new cannot
# delete a session while an earlier response is still updating it.
async with session_locks.setdefault(session_id, asyncio.Lock()):
if (command := telegram_command(update)) is not None and await handle_command(update, command):
return
async def resolve_file_url(file_id: str) -> str | None:
file = await bot.get_file(file_id)
if file.file_path is None or (file.file_size is not None and file.file_size > MAX_MEDIA_BYTES):
return None
destination = BytesIO()
await bot.download_file(file.file_path, destination=destination)
data = destination.getvalue()
if len(data) > MAX_MEDIA_BYTES:
return None
encoded = base64.b64encode(data).decode("ascii")
return f"data:application/octet-stream;base64,{encoded}"
try:
run = await telegram_to_run(update, resolve_file_url=resolve_file_url, stream=True)
except ValueError:
LOGGER.debug("Ignoring non-actionable Telegram update", exc_info=True)
return
await bot.send_chat_action(chat_id=chat_id, action="typing")
placeholder = await bot.send_message(chat_id=chat_id, text=PLACEHOLDER_TEXT)
target = await state.get_target()
# Reuse one AgentSession per Telegram chat. The /new command removes this
# mapping so get_or_create_session creates a clean session next time.
session = await state.get_or_create_session(session_id)
stream = target.run(
run["messages"],
stream=True,
session=session,
options=run["options"],
)
if not isinstance(stream, ResponseStream):
raise RuntimeError("agent did not return a response stream")
last_edit_at = 0.0
async for operation in telegram_from_streaming_run(
stream,
chat_id=chat_id,
message_id=placeholder.message_id,
initial_text=PLACEHOLDER_TEXT,
):
if operation["method"] == "editMessageText":
delay = EDIT_INTERVAL_SECONDS - (time.monotonic() - last_edit_at)
if delay > 0:
await asyncio.sleep(delay)
last_edit_at = time.monotonic()
await execute_operation(operation)
# Persist the updated AgentSession back under the stable per-chat key after
# streaming has finalized and the history provider has recorded the turn.
await state.set_session(session_id, session)
ポーリングと Webhook のセットアップ、コマンド処理、受信メディア ポリシー、ストリーミング編集、運用展開のガイダンスについては、 ローカルの Telegram サンプルを参照してください。
Important
更新を処理する前に、Telegram Webhook の配信を確認します。 Webhook シークレットは Telegram の配信を認証しますが、Telegram ユーザーまたはチャットがアプリケーション データにアクセスすることを承認しません。 アプリケーションが承認ポリシーを適用するまで、チャット ID とユーザー ID を信頼されていないものとして扱います。
次のステップ
より深く進む: