비고
.NET용 텔레그램 봇을 위한 자체 호스팅용 도우미가 곧 제공될 예정입니다.
비고
Telegram 봇에 대한 자체 호스팅 도우미는 현재 Go에서 사용할 수 없습니다.
agent-framework-hosting-telegram 는 Telegram Bot API 업데이트를 Agent Framework 실행 값으로 변환하고 최종 또는 스트리밍 실행을 Bot API 작업으로 렌더링합니다. 봇 클라이언트, 폴링 런타임, 웹후크 라우터, 명령 레지스트리 또는 배달 프레임워크를 제공하지 않습니다.
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-telegram azure-identity
업데이트 페이로드를 제공하고 도우미가 반환한 작업을 실행할 수 있는 모든 Telegram 클라이언트 라이브러리를 사용합니다. 샘플은 aiogram를 사용하지만, 도우미는 이에 종속되지 않습니다.
업데이트 처리
aiogram 웹후크 샘플은 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:
username = (await bot.me()).username
if not username or telegram_command(update, bot_username=username) is None:
return
if 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)
명령을 디스패치하기 전에 현재 봇 사용자 이름을 전달합니다 telegram_command(update, bot_username=...). 헬퍼는 메시지 텍스트, 콜백 데이터, 미디어 캡션에 있는 명령을 포함해 다른 봇을 대상으로 한 명령에 대해 None를 반환합니다.
bot_username를 생략하면 호환성을 유지하기 위해 대상 독립적 구문 분석이 유지됩니다.
폴링 및 웹후크 설정, 명령 처리, 인바운드 미디어 정책, 스트리밍 편집 및 프로덕션 배포 지침은 로컬 Telegram 샘플을 참조하세요.
중요합니다
업데이트를 처리하기 전에 Telegram 웹후크 배달을 확인합니다. 웹후크 암호는 Telegram의 배달을 인증하지만 애플리케이션 데이터에 액세스하도록 Telegram 사용자 또는 채팅에 권한을 부여하지는 않습니다. 애플리케이션이 권한 부여 정책을 적용할 때까지 채팅 및 사용자 ID를 신뢰할 수 없는 것으로 처리합니다.
다음 단계
더 자세히 살펴보기: