From 18a7a75bacc69c0674a08ca2cf81212ac8f2d834 Mon Sep 17 00:00:00 2001 From: CatalanCabbage Date: Wed, 14 Oct 2026 02:06:32 +0530 Subject: [PATCH 1/5] Add Telegram channel adapter Long polling by default with offset holdback, optional secret-checked webhook mode, Markdown converted to Telegram entities, typing keepalive, inline-keyboard rich replies, document uploads and bounded downloads. --- pyproject.toml | 1 + src/agentchat/channels/telegram/__init__.py | 27 + src/agentchat/channels/telegram/adapter.py | 501 +++++++++++++++++++ src/agentchat/channels/telegram/events.py | 221 ++++++++ src/agentchat/channels/telegram/rendering.py | 216 ++++++++ src/agentchat/channels/telegram/transport.py | 146 ++++++ 6 files changed, 1112 insertions(+) create mode 100644 src/agentchat/channels/telegram/__init__.py create mode 100644 src/agentchat/channels/telegram/adapter.py create mode 100644 src/agentchat/channels/telegram/events.py create mode 100644 src/agentchat/channels/telegram/rendering.py create mode 100644 src/agentchat/channels/telegram/transport.py diff --git a/pyproject.toml b/pyproject.toml index 385d9a2..aa84c5d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ teams = [ ] imessage = ["python-socketio[asyncio-client]>=5.13,<6"] whatsapp = [] # Uses the existing aiohttp dependency. +telegram = [] # Uses the existing aiohttp dependency. dev = [ "openai-agents>=0.4,<1", "discord.py>=2.6,<3", diff --git a/src/agentchat/channels/telegram/__init__.py b/src/agentchat/channels/telegram/__init__.py new file mode 100644 index 0000000..3ef36c4 --- /dev/null +++ b/src/agentchat/channels/telegram/__init__.py @@ -0,0 +1,27 @@ +"""Telegram adapter and provider-local helpers, loaded on demand.""" + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from agentchat.channels.telegram.adapter import Telegram + +__all__ = ["Telegram"] + + +def __getattr__(name: str) -> object: + if name != "Telegram": + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + try: + from agentchat.channels.telegram.adapter import Telegram + except ModuleNotFoundError as exc: + if exc.name and exc.name.split(".")[0] == "aiohttp": + raise ModuleNotFoundError( + "Install Telegram support with: pip install 'agentchat[telegram]'", + name=exc.name, + ) from exc + raise + return Telegram + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(__all__)) diff --git a/src/agentchat/channels/telegram/adapter.py b/src/agentchat/channels/telegram/adapter.py new file mode 100644 index 0000000..4d88808 --- /dev/null +++ b/src/agentchat/channels/telegram/adapter.py @@ -0,0 +1,501 @@ +from __future__ import annotations + +import asyncio +import contextlib +import hmac +import json +import logging +import os +import posixpath +import re +from collections.abc import Mapping, Sequence +from types import MappingProxyType +from typing import Any, NamedTuple, Protocol +from urllib.parse import urlsplit + +from aiohttp import web + +from agentchat._delivery import DeliveryTasks +from agentchat.channels.base import MessageReceiver +from agentchat.channels.telegram.events import normalize +from agentchat.channels.telegram.rendering import text_payload, truncate_caption +from agentchat.channels.telegram.transport import ( + REQUEST_TIMEOUT, + TelegramBotClient, + TelegramError, +) +from agentchat.models import ( + Attachment, + DownloadedFile, + Message, + RichReply, + Sender, + UploadedFile, + UploadFile, +) + +logger = logging.getLogger(__name__) + +POLL_TIMEOUT = 30 +# The HTTP timeout must outlive Telegram's long-poll hold or every empty poll times out. +POLL_HTTP_MARGIN = 10 +ALLOWED_UPDATES = ["message", "callback_query"] +WEBHOOK_PATH = "/webhooks/telegram" +SECRET_HEADER = "X-Telegram-Bot-Api-Secret-Token" +RETRY_BASE = 1.0 +RETRY_MAX = 30.0 +BUSY_DELAY = 1.0 +CONFIRM_TIMEOUT = 2 +ANSWER_TIMEOUT = 5 +# Telegram clears a chat action after about five seconds. +TYPING_INTERVAL = 4 +TYPING_EXPIRY = 120 +BOT_TOKEN_PATTERN = re.compile(r"([0-9]+):[A-Za-z0-9_-]+") +WEBHOOK_SECRET_PATTERN = re.compile(r"[A-Za-z0-9_-]{1,256}") +CALLBACK_ONLY_METADATA = frozenset({"callback_query_id", "callback_data"}) + + +class BotAPIClient(Protocol): + async def call( + self, + method: str, + payload: dict | None = None, + *, + request_timeout: float = REQUEST_TIMEOUT, + ) -> Any: ... + + async def upload( + self, *, method: str, fields: dict, file_field: str, filename: str, content: bytes, + ) -> Any: ... + + async def download_file(self, file_id: str, *, max_bytes: int) -> tuple[str, bytes]: ... + + async def close(self) -> None: ... + + +class _Destination(NamedTuple): + chat_id: int + message_id: int + thread_id: int | None + + +def retry_delay(attempt: int, retry_after: int | None) -> float: + if retry_after: + return float(retry_after) + return min(RETRY_BASE * 2**attempt, RETRY_MAX) + + +def _is_plain_filename(filename: object) -> bool: + return ( + isinstance(filename, str) + and bool(filename) + and len(filename.encode()) <= 255 + and filename not in {".", ".."} + and not any(char in filename for char in "/\\") + and not any(ord(char) < 32 for char in filename) + ) + + +class Telegram: + """Telegram Bot API: long polling by default, or a secret-checked webhook.""" + + name = "telegram" + + def __init__( + self, + *, + bot_token: str, + webhook_url: str | None = None, + webhook_secret: str | None = None, + host: str = "127.0.0.1", + port: int = 3980, + client: BotAPIClient | None = None, + max_pending_events: int = 64, + shutdown_timeout: float = 20, + ) -> None: + if not isinstance(bot_token, str) or not bot_token.strip(): + raise ValueError("Telegram bot_token is required") + token_match = BOT_TOKEN_PATTERN.fullmatch(bot_token) + if token_match is None: + raise ValueError("Telegram bot_token must look like 123456:ABC...") + if (webhook_url is None) != (webhook_secret is None): + raise ValueError("Telegram webhook_url and webhook_secret must be given together") + if webhook_url is not None: + target = urlsplit(webhook_url) + if target.scheme != "https" or not target.hostname or target.path != WEBHOOK_PATH: + raise ValueError(f"Telegram webhook_url must be https and end in {WEBHOOK_PATH}") + if webhook_secret is not None and not WEBHOOK_SECRET_PATTERN.fullmatch(webhook_secret): + raise ValueError("Telegram webhook_secret must be 1-256 of A-Z, a-z, 0-9, _ and -") + self._bot_id = int(token_match[1]) + self._username: str | None = None + self._first_name: str | None = None + self._webhook_url, self._webhook_secret = webhook_url, webhook_secret + self._host, self._port = host, port + self._client = client or TelegramBotClient(bot_token) + self._delivery = DeliveryTasks(max_pending_events, shutdown_timeout) + self._offset: int | None = None + self._confirmed_offset: int | None = None + self._worker: asyncio.Task[None] | None = None + self._runner: web.AppRunner | None = None + self._typing: dict[tuple[int, int | None], asyncio.Task[None]] = {} + self._answers: set[asyncio.Task[None]] = set() + self._close_lock = asyncio.Lock() + self._closed = False + + @classmethod + def from_env(cls) -> Telegram: + token = os.getenv("TELEGRAM_BOT_TOKEN") + if not token: + raise RuntimeError("TELEGRAM_BOT_TOKEN is required") + webhook_url = os.getenv("TELEGRAM_WEBHOOK_URL") or None + webhook_secret = os.getenv("TELEGRAM_WEBHOOK_SECRET") or None + if (webhook_url is None) != (webhook_secret is None): + raise RuntimeError( + "TELEGRAM_WEBHOOK_URL and TELEGRAM_WEBHOOK_SECRET are required together" + ) + return cls( + bot_token=token, + webhook_url=webhook_url, + webhook_secret=webhook_secret, + host=os.getenv("TELEGRAM_HOST", "127.0.0.1"), + port=int(os.getenv("TELEGRAM_PORT", "3980")), + ) + + def bind(self, receiver: MessageReceiver) -> None: + self._delivery.receiver = receiver + + def create_app(self) -> web.Application: + if self._webhook_secret is None: + raise RuntimeError("Telegram webhooks need webhook_url and webhook_secret") + app = web.Application() + app.router.add_post(WEBHOOK_PATH, self._receive) + return app + + async def run(self) -> None: + self._delivery.check_bound() + if self._worker is not None: + raise RuntimeError("Telegram channel is already running") + self._worker = asyncio.create_task(self._serve()) + try: + await asyncio.wait({self._worker}) + finally: + await self.close() + if not self._worker.cancelled(): + self._worker.result() + + async def close(self) -> None: + async with self._close_lock: + if self._closed: + return + self._closed = True + self._delivery.closed = True + try: + worker = self._worker + if worker is not None and worker is not asyncio.current_task(): + worker.cancel() + await asyncio.wait({worker}) + if self._runner: + await self._runner.cleanup() + await self._confirm_offset() + finally: + try: + await self._delivery.close() + finally: + try: + background = {*self._typing.values(), *self._answers} + for task in background: + task.cancel() + await asyncio.gather(*background, return_exceptions=True) + finally: + await self._client.close() + + async def _identify(self) -> None: + try: + me = await self._client.call("getMe") + except TelegramError as error: + if error.is_transient: + raise + raise TelegramError( + "Telegram bot authentication failed", error_code=error.error_code, + ) from None + username = me.get("username") if isinstance(me, dict) else None + if not isinstance(username, str) or not username: + raise RuntimeError("Telegram did not return the bot identity") + if me.get("id") != self._bot_id: + raise RuntimeError("Telegram token does not match its bot identity") + first_name = me.get("first_name") + self._first_name = first_name if isinstance(first_name, str) else username + self._username = username + + async def _serve(self) -> None: + await self._identify() + if self._webhook_url is None: + await self._client.call("deleteWebhook", {"drop_pending_updates": False}) + await self._poll() + else: + await self._serve_webhook() + + async def _poll(self) -> None: + attempt = 0 + while True: + payload: dict[str, object] = { + "timeout": POLL_TIMEOUT, "allowed_updates": ALLOWED_UPDATES, + } + if self._offset is not None: + payload["offset"] = self._offset + try: + updates = await self._client.call( + "getUpdates", payload, request_timeout=POLL_TIMEOUT + POLL_HTTP_MARGIN + ) + except TelegramError as error: + if not error.is_transient: + raise + delay = retry_delay(attempt, error.retry_after) + logger.warning("Telegram polling failed (%s); retrying in %.1fs", error, delay) + attempt += 1 + await asyncio.sleep(delay) + continue + attempt = 0 + self._confirmed_offset = self._offset + if not self._admit(updates if isinstance(updates, list) else []): + await asyncio.sleep(BUSY_DELAY) + + def _admit(self, updates: list) -> bool: + for update in updates: + update_id = update.get("update_id") if isinstance(update, dict) else None + if not isinstance(update_id, int): + continue + message = self._normalize(update) + if message is not None and not self._submit(message): + return False + self._offset = update_id + 1 + return True + + def _normalize(self, update: dict) -> Message | None: + assert self._username is not None + return normalize(update, bot_id=self._bot_id, bot_username=self._username) + + def _submit(self, message: Message) -> bool: + if not self._delivery.submit(self, message): + return False + callback_id = message.metadata.get("callback_query_id") + if isinstance(callback_id, str): + task = asyncio.create_task(self._answer_callback(callback_id)) + self._answers.add(task) + task.add_done_callback(self._answers.discard) + return True + + async def _answer_callback(self, callback_id: str) -> None: + try: + await self._client.call( + "answerCallbackQuery", + {"callback_query_id": callback_id}, + request_timeout=ANSWER_TIMEOUT, + ) + except Exception as error: + logger.warning("Telegram callback answer failed (%s)", error) + + async def _confirm_offset(self) -> None: + # Accepted updates stay unconfirmed until the next getUpdates; a restart would replay them. + if self._offset is None or self._offset == self._confirmed_offset: + return + try: + await self._client.call( + "getUpdates", + {"offset": self._offset, "limit": 1, "timeout": 0}, + request_timeout=CONFIRM_TIMEOUT, + ) + except TelegramError as error: + logger.warning("Telegram offset confirmation failed (%s)", error) + + async def _serve_webhook(self) -> None: + self._runner = web.AppRunner(self.create_app(), access_log=None) + await self._runner.setup() + await web.TCPSite(self._runner, self._host, self._port).start() + # Listen first: Telegram starts sending as soon as setWebhook succeeds. + await self._client.call( + "setWebhook", + { + "url": self._webhook_url, + "secret_token": self._webhook_secret, + "allowed_updates": ALLOWED_UPDATES, + }, + ) + await asyncio.Event().wait() + + async def _receive(self, request: web.Request) -> web.Response: + assert self._webhook_secret is not None + supplied = request.headers.get(SECRET_HEADER, "") + if not hmac.compare_digest(supplied.encode(), self._webhook_secret.encode()): + raise web.HTTPUnauthorized(text="Invalid webhook secret") + try: + update = json.loads(await request.read()) + except (ValueError, UnicodeError): + raise web.HTTPBadRequest(text="Invalid webhook JSON") from None + if not isinstance(update, dict) or not isinstance(update.get("update_id"), int): + raise web.HTTPBadRequest(text="Invalid webhook update") + if self._delivery.closed: + raise web.HTTPServiceUnavailable(text="Telegram receiver is closed") + if self._username is None: + try: + await self._identify() + except Exception as error: + logger.warning("Telegram bot identity unavailable (%s)", error) + raise web.HTTPServiceUnavailable(text="Telegram receiver is starting") from None + message = self._normalize(update) + if message is not None and not self._submit(message): + raise web.HTTPServiceUnavailable(text="Telegram receiver is busy") + return web.Response(text="OK") + + def _destination(self, source: Message) -> _Destination: + data = source.metadata + chat_id, message_id = data.get("chat_id"), data.get("message_id") + thread = data.get("message_thread_id") + if ( + source.channel != self.name + or data.get("bot_id") != self._bot_id + or not isinstance(chat_id, int) + or not isinstance(message_id, int) + or isinstance(chat_id, bool) + or isinstance(message_id, bool) + or (thread is not None and (not isinstance(thread, int) or isinstance(thread, bool))) + ): + raise ValueError("Source does not belong to this Telegram adapter") + return _Destination(chat_id, message_id, thread) + + async def _send( + self, *, source: Message, destination: _Destination, payload: dict, text: str, + ) -> Message: + self._delivery.check_reply() + payload = { + **payload, + "chat_id": destination.chat_id, + "reply_parameters": { + "message_id": destination.message_id, "allow_sending_without_reply": True, + }, + } + if destination.thread_id is not None: + payload["message_thread_id"] = destination.thread_id + result = await self._client.call("sendMessage", payload) + sent = result.get("message_id") if isinstance(result, dict) else None + if not isinstance(sent, int) or isinstance(sent, bool) or sent <= 0: + raise RuntimeError("Telegram did not return a message receipt") + metadata = { + key: value for key, value in source.metadata.items() + if key not in CALLBACK_ONLY_METADATA + } + return Message( + id=f"telegram:{self._bot_id}:{destination.chat_id}:{sent}", + conversation_id=source.conversation_id, + channel=self.name, + sender=Sender(id=str(self._bot_id), display_name=self._first_name), + text=text, + role="assistant", + thread_id=source.thread_id, + metadata=MappingProxyType({**metadata, "message_id": sent}), + ) + + async def reply(self, source: Message, content: str) -> Message: + destination = self._destination(source) + return await self._send( + source=source, destination=destination, payload=text_payload(content), text=content, + ) + + async def reply_rich(self, source: Message, content: RichReply) -> Message: + destination = self._destination(source) + if content.attachments: + raise ValueError("Telegram rich replies do not support attachments") + if len(content.blocks) != 1 or not isinstance(content.blocks[0], Mapping): + raise ValueError("Telegram rich replies need exactly one reply_markup mapping") + payload = {**text_payload(content.text), "reply_markup": dict(content.blocks[0])} + return await self._send( + source=source, destination=destination, payload=payload, text=content.text, + ) + + async def set_status(self, source: Message, status: str) -> bool: + destination = self._destination(source) + self._delivery.check_reply() + key = (destination.chat_id, destination.thread_id) + if previous := self._typing.pop(key, None): + previous.cancel() + if not status: + return True + payload: dict[str, object] = {"chat_id": destination.chat_id, "action": "typing"} + if destination.thread_id is not None: + payload["message_thread_id"] = destination.thread_id + await self._client.call("sendChatAction", payload) + # A concurrent call for the same chat may have stored a task while this one awaited. + if previous := self._typing.pop(key, None): + previous.cancel() + self._typing[key] = asyncio.create_task(self._keep_typing(key, payload)) + return True + + async def _keep_typing(self, key: tuple[int, int | None], payload: dict) -> None: + loop = asyncio.get_running_loop() + deadline = loop.time() + TYPING_EXPIRY + try: + while loop.time() + TYPING_INTERVAL < deadline: + await asyncio.sleep(TYPING_INTERVAL) + with contextlib.suppress(Exception): + await self._client.call( + "sendChatAction", payload, request_timeout=TYPING_INTERVAL, + ) + finally: + if self._typing.get(key) is asyncio.current_task(): + del self._typing[key] + + async def upload_files( + self, source: Message, files: Sequence[UploadFile], + ) -> tuple[UploadedFile, ...]: + destination = self._destination(source) + batch = tuple(files) + if not batch: + raise ValueError("Expected at least one file") + for file in batch: + if not _is_plain_filename(file.filename): + raise ValueError("Expected a plain filename") + if not isinstance(file.content, bytes) or not file.content: + raise ValueError("Expected nonempty file bytes") + if file.title is not None and not isinstance(file.title, str): + raise ValueError("Expected a file title") + uploaded = [] + for file in batch: + self._delivery.check_reply() + fields: dict[str, object] = {"chat_id": destination.chat_id} + if destination.thread_id is not None: + fields["message_thread_id"] = destination.thread_id + if file.title: + fields["caption"] = truncate_caption(file.title) + result = await self._client.upload( + method="sendDocument", + fields=fields, + file_field="document", + filename=file.filename, + content=file.content, + ) + document = result.get("document") if isinstance(result, dict) else None + file_id = document.get("file_id") if isinstance(document, dict) else None + if not isinstance(file_id, str) or not file_id: + raise RuntimeError("Telegram did not return a document receipt") + uploaded.append(UploadedFile(id=file_id)) + return tuple(uploaded) + + async def download_attachment( + self, source: Message, attachment: Attachment, *, max_bytes: int = 10 * 1024 * 1024, + ) -> DownloadedFile: + if source.role != "user": + raise ValueError("Source does not belong to this Telegram adapter") + self._destination(source) + if attachment not in source.attachments: + raise ValueError("Attachment does not belong to the source message") + if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes <= 0: + raise ValueError("Expected a positive attachment byte limit") + path, content = await self._client.download_file(attachment.id, max_bytes=max_bytes) + if not content: + raise ValueError("This Telegram file was empty. Send the attachment again.") + return DownloadedFile( + id=attachment.id, + name=attachment.name or posixpath.basename(path) or "attachment", + content=content, + media_type=attachment.media_type, + ) diff --git a/src/agentchat/channels/telegram/events.py b/src/agentchat/channels/telegram/events.py new file mode 100644 index 0000000..ab687b4 --- /dev/null +++ b/src/agentchat/channels/telegram/events.py @@ -0,0 +1,221 @@ +"""Normalize Telegram Bot API updates; no network calls.""" + +from __future__ import annotations + +from types import MappingProxyType + +from agentchat.models import Attachment, Message, Sender + +CHAT_PRIVATE = "private" +GROUP_CHATS = frozenset({"group", "supergroup"}) +ENTITY_MENTION = "mention" +ENTITY_COMMAND = "bot_command" +ENTITY_TEXT_MENTION = "text_mention" +PHOTO_MEDIA_TYPE = "image/jpeg" +# Animations also arrive as a document, so the earlier kind wins for a shared file_id. +FILE_KINDS = ("animation", "document", "audio", "video", "voice", "video_note", "sticker") + + +def _mapping(value: object) -> dict: + return value if isinstance(value, dict) else {} + + +def _items(value: object) -> list: + return value if isinstance(value, list) else [] + + +def _is_id(value: object) -> bool: + return isinstance(value, int) and not isinstance(value, bool) + + +def _sender(user: dict) -> Sender | None: + if not _is_id(user.get("id")) or user.get("is_bot") is not False: + return None + names = (user.get("first_name"), user.get("last_name")) + display_name = " ".join(name for name in names if isinstance(name, str) and name) + return Sender(id=str(user["id"]), display_name=display_name or None) + + +def _span(units: bytes, entity: dict) -> tuple[int, int, str] | None: + offset, length = entity.get("offset"), entity.get("length") + if not _is_id(offset) or not _is_id(length) or offset < 0 or length <= 0: + return None + end = offset + length + if end * 2 > len(units): + return None + try: + return offset, end, units[offset * 2:end * 2].decode("utf-16-le") + except UnicodeDecodeError: + return None + + +def _address( + *, text: str, entities: list, bot_id: int, bot_username: str, +) -> tuple[bool, str]: + units = text.encode("utf-16-le") + mention = "@" + bot_username.lower() + addressed, removals = False, [] + for raw in entities: + entity = _mapping(raw) + span = _span(units, entity) + if span is None: + continue + start, end, value = span + kind = entity.get("type") + if kind == ENTITY_MENTION and value.lower() == mention: + addressed = True + removals.append((start, end)) + elif kind == ENTITY_COMMAND and value.lower().endswith(mention): + addressed = True + removals.append((end - len(mention), end)) + elif kind == ENTITY_TEXT_MENTION and _mapping(entity.get("user")).get("id") == bot_id: + addressed = True + kept, cursor = [], 0 + for start, end in sorted(removals): + if start < cursor: + continue + kept.append(units[cursor * 2:start * 2]) + cursor = end + kept.append(units[cursor * 2:]) + return addressed, b"".join(kept).decode("utf-16-le").strip() + + +def _attachments(message: dict) -> tuple[Attachment, ...]: + found: dict[str, Attachment] = {} + photos = [item for item in _items(message.get("photo")) if isinstance(item, dict)] + if photos and isinstance(photos[-1].get("file_id"), str) and photos[-1]["file_id"]: + size = photos[-1].get("file_size") + found[photos[-1]["file_id"]] = Attachment( + id=photos[-1]["file_id"], + media_type=PHOTO_MEDIA_TYPE, + size=size if _is_id(size) else None, + ) + for kind in FILE_KINDS: + media = _mapping(message.get(kind)) + file_id = media.get("file_id") + if not isinstance(file_id, str) or not file_id or file_id in found: + continue + name, media_type = media.get("file_name"), media.get("mime_type") + size = media.get("file_size") + found[file_id] = Attachment( + id=file_id, + name=name if isinstance(name, str) and name else None, + media_type=media_type if isinstance(media_type, str) and media_type else None, + size=size if _is_id(size) else None, + ) + return tuple(found.values()) + + +def _button_label(message: dict, data: str) -> str: + keyboard = _mapping(message.get("reply_markup")).get("inline_keyboard") + for row in _items(keyboard): + for button in _items(row): + button = _mapping(button) + if button.get("callback_data") == data and isinstance(button.get("text"), str): + return button["text"] + return data + + +def _message( + *, bot_id: int, local_id: str, chat: dict, message: dict, sender: Sender, text: str, + attachments: tuple[Attachment, ...] = (), extra: dict | None = None, +) -> Message: + thread = message.get("message_thread_id") + forum_topic = thread if message.get("is_topic_message") is True and _is_id(thread) else None + conversation_id = f"telegram:{bot_id}:{chat['id']}" + return Message( + id=f"telegram:{bot_id}:{local_id}", + conversation_id=( + conversation_id if forum_topic is None else f"{conversation_id}:{forum_topic}" + ), + channel="telegram", + sender=sender, + text=text, + role="user", + thread_id=None if forum_topic is None else str(forum_topic), + attachments=attachments, + metadata=MappingProxyType( + { + "bot_id": bot_id, + "chat_id": chat["id"], + "chat_type": chat["type"], + "message_id": message["message_id"], + "message_thread_id": forum_topic, + **(extra or {}), + } + ), + ) + + +def _chat(message: dict) -> dict | None: + chat = _mapping(message.get("chat")) + message_id = message.get("message_id") + if ( + not _is_id(chat.get("id")) + or (chat.get("type") != CHAT_PRIVATE and chat.get("type") not in GROUP_CHATS) + or not _is_id(message_id) + or message_id <= 0 + ): + return None + return chat + + +def _normalize_callback(callback: dict, *, bot_id: int) -> Message | None: + message = _mapping(callback.get("message")) + data, callback_id = callback.get("data"), callback.get("id") + sender = _sender(_mapping(callback.get("from"))) + chat = _chat(message) + if ( + not isinstance(data, str) + or not data + or not isinstance(callback_id, str) + or not callback_id + or sender is None + or chat is None + or message.get("date") == 0 + ): + return None + return _message( + bot_id=bot_id, + local_id=f"callback:{callback_id}", + chat=chat, + message=message, + sender=sender, + text=_button_label(message, data), + extra={"callback_query_id": callback_id, "callback_data": data}, + ) + + +def normalize(update: object, *, bot_id: int, bot_username: str) -> Message | None: + update = _mapping(update) + if "callback_query" in update: + return _normalize_callback(_mapping(update["callback_query"]), bot_id=bot_id) + message = _mapping(update.get("message")) + sender = _sender(_mapping(message.get("from"))) + chat = _chat(message) + if sender is None or chat is None: + return None + if isinstance(message.get("text"), str): + text, entities = message["text"], _items(message.get("entities")) + elif isinstance(message.get("caption"), str): + text, entities = message["caption"], _items(message.get("caption_entities")) + else: + text, entities = "", [] + addressed, text = _address( + text=text, entities=entities, bot_id=bot_id, bot_username=bot_username, + ) + reply_to = _mapping(_mapping(message.get("reply_to_message")).get("from")) + if chat["type"] != CHAT_PRIVATE and not addressed and reply_to.get("id") != bot_id: + return None + attachments = _attachments(message) + if not text and not attachments: + return None + return _message( + bot_id=bot_id, + local_id=f"{chat['id']}:{message['message_id']}", + chat=chat, + message=message, + sender=sender, + text=text, + attachments=attachments, + ) diff --git a/src/agentchat/channels/telegram/rendering.py b/src/agentchat/channels/telegram/rendering.py new file mode 100644 index 0000000..2a51a85 --- /dev/null +++ b/src/agentchat/channels/telegram/rendering.py @@ -0,0 +1,216 @@ +import re +import string + +TELEGRAM_TEXT_LIMIT = 4096 +TELEGRAM_CAPTION_LIMIT = 1024 + +_BOLD = "bold" +_ITALIC = "italic" +_CODE = "code" +_PRE = "pre" +_STRIKETHROUGH = "strikethrough" +_SPOILER = "spoiler" +_BLOCKQUOTE = "blockquote" +_TEXT_LINK = "text_link" + +_FENCE = re.compile(r"[ \t]*(`{3,}|~{3,})[ \t]*([^\n`]*)") +_HEADING = re.compile(r"#{1,6}\s+(.*)") +_BULLET = re.compile(r"(\s*)[-*+]\s+(.*)") +_NUMBERED = re.compile(r"(\s*)(\d+)\.\s+(.*)") +_QUOTE = re.compile(r">\s?(.*)") +# Telegram rejects the whole message when a text_link url fails its link check. +_LINK = re.compile(r'\[([^\]\n]*)\]\(((?:https?|tg)://[^)\s]+)(?:\s+"[^"]*")?\)') +_WORD_CHAR = re.compile(r"\w", re.ASCII) +_ESCAPABLE = frozenset(string.punctuation) + +_INLINE = ( + ("```", _CODE), + ("**", _BOLD), + ("~~", _STRIKETHROUGH), + ("||", _SPOILER), + ("`", _CODE), + ("*", _ITALIC), + ("_", _ITALIC), +) +_INTRAWORD_LITERAL = frozenset({"*", "_"}) + + +def _utf16_len(text: str) -> int: + return len(text) + sum(1 for char in text if ord(char) > 0xFFFF) + + +class _Builder: + def __init__(self) -> None: + self.parts: list[str] = [] + self.length = 0 + self.entities: list[dict] = [] + + def add(self, segment: str, entity_type: str | None = None, **extra: str) -> None: + if not segment: + return + size = _utf16_len(segment) + if entity_type: + self.entities.append( + {"type": entity_type, "offset": self.length, "length": size, **extra} + ) + self.parts.append(segment) + self.length += size + + def wrap(self, start: int, entity_type: str, **extra: str) -> None: + if self.length > start: + self.entities.append( + {"type": entity_type, "offset": start, "length": self.length - start, **extra} + ) + + +def _is_word_char(src: str, index: int) -> bool: + return 0 <= index < len(src) and bool(_WORD_CHAR.match(src[index])) + + +def _render_inline(b: _Builder, src: str, active: frozenset[str] = frozenset()) -> None: + i = 0 + plain: list[str] = [] + + def flush() -> None: + if plain: + b.add("".join(plain)) + plain.clear() + + while i < len(src): + if src[i] == "\\" and i + 1 < len(src) and src[i + 1] in _ESCAPABLE: + plain.append(src[i + 1]) + i += 2 + continue + + link = _LINK.match(src, i) + if link: + flush() + start = b.length + _render_inline(b, link[1], active | {_TEXT_LINK}) + b.wrap(start, _TEXT_LINK, url=link[2]) + i = link.end() + continue + + for marker, entity_type in _INLINE: + if not src.startswith(marker, i) or entity_type in active: + continue + intraword_literal = marker in _INTRAWORD_LITERAL + if intraword_literal and _is_word_char(src, i - 1): + continue + start_inner = i + len(marker) + end = src.find(marker, start_inner) + if end in (-1, start_inner): + continue + if intraword_literal and _is_word_char(src, end + len(marker)): + continue + inner = src[start_inner:end] + if entity_type != _CODE and (inner[0].isspace() or inner[-1].isspace()): + continue + flush() + if entity_type == _CODE: + b.add(inner, None if _TEXT_LINK in active else _CODE) + else: + start = b.length + _render_inline(b, inner, active | {entity_type}) + b.wrap(start, entity_type) + i = end + len(marker) + break + else: + plain.append(src[i]) + i += 1 + flush() + + +def _render_line(b: _Builder, line: str) -> None: + if heading := _HEADING.fullmatch(line): + start = b.length + _render_inline(b, heading[1]) + b.wrap(start, _BOLD) + elif bullet := _BULLET.fullmatch(line): + b.add(f"{bullet[1]}โ€ข ") + _render_inline(b, bullet[2]) + elif numbered := _NUMBERED.fullmatch(line): + b.add(f"{numbered[1]}{numbered[2]}. ") + _render_inline(b, numbered[3]) + else: + _render_inline(b, line) + + +def _render_block(b: _Builder, block: list[str]) -> None: + quote_start: int | None = None + for n, line in enumerate(block): + quote = _QUOTE.fullmatch(line) + if quote_start is not None and not quote: + b.wrap(quote_start, _BLOCKQUOTE) + quote_start = None + if n: + b.add("\n") + if quote: + if quote_start is None: + quote_start = b.length + _render_inline(b, quote[1]) + else: + _render_line(b, line) + if quote_start is not None: + b.wrap(quote_start, _BLOCKQUOTE) + + +def render(markdown: str) -> tuple[str, list[dict]]: + b = _Builder() + lines = markdown.replace("\r\n", "\n").split("\n") + i = 0 + while i < len(lines): + if fence := _FENCE.fullmatch(lines[i]): + closer = fence[1][0] * 3 + language = fence[2].split()[0] if fence[2].strip() else None + body: list[str] = [] + i += 1 + while i < len(lines) and not lines[i].lstrip().startswith(closer): + body.append(lines[i]) + i += 1 + i += 1 + code = "\n".join(body).rstrip("\n") + if code: + if b.length: + b.add("\n\n") + b.add(code, _PRE, **({"language": language} if language else {})) + continue + + if not lines[i].strip(): + i += 1 + continue + + block: list[str] = [] + while i < len(lines) and lines[i].strip() and not _FENCE.fullmatch(lines[i]): + block.append(lines[i]) + i += 1 + if b.length: + b.add("\n\n") + _render_block(b, block) + + # Wrappers are appended after their contents, so equal spans list the outer entity first. + order = sorted( + range(len(b.entities)), + key=lambda n: (b.entities[n]["offset"], -b.entities[n]["length"], -n), + ) + return "".join(b.parts), [b.entities[n] for n in order] + + +def text_payload(content: str) -> dict: + if not isinstance(content, str) or not content.strip(): + raise ValueError("Telegram replies require nonempty text") + text, entities = render(content) + if not text.strip(): + raise ValueError("Telegram replies require nonempty text") + if _utf16_len(text) > TELEGRAM_TEXT_LIMIT: + raise ValueError("Telegram replies must fit 4,096 characters; send separate replies") + payload: dict = {"text": text} + if entities: + payload["entities"] = entities + payload["link_preview_options"] = {"is_disabled": True} + return payload + + +def truncate_caption(title: str) -> str: + units = title.encode("utf-16-le")[: TELEGRAM_CAPTION_LIMIT * 2] + return units.decode("utf-16-le", errors="ignore") diff --git a/src/agentchat/channels/telegram/transport.py b/src/agentchat/channels/telegram/transport.py new file mode 100644 index 0000000..3304d13 --- /dev/null +++ b/src/agentchat/channels/telegram/transport.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import json +from typing import Any +from urllib.parse import quote + +import aiohttp + +API_ROOT = "https://api.telegram.org" +REQUEST_TIMEOUT = 30 +UPLOAD_TIMEOUT = 60 +DOWNLOAD_TIMEOUT = 60 +DOWNLOAD_CHUNK = 64 * 1024 + + +def _download_limit_error(max_bytes: int) -> ValueError: + return ValueError(f"Telegram file exceeds the {max_bytes / (1024 * 1024):g} MB download limit") + + +class TelegramError(RuntimeError): + def __init__( + self, + description: str, + *, + error_code: int | None = None, + retry_after: int | None = None, + ) -> None: + super().__init__(description) + self.description = description + self.error_code = error_code + self.retry_after = retry_after + + @property + def is_transient(self) -> bool: + code = self.error_code + return code is None or code == 429 or code >= 500 + + +def _unwrap_result(method: str, status: int, body: object) -> Any: + if isinstance(body, dict) and body.get("ok") is True and "result" in body: + return body["result"] + if not isinstance(body, dict): + raise TelegramError(f"Telegram {method} failed (HTTP {status})", error_code=status) + parameters = body.get("parameters") + retry_after = parameters.get("retry_after") if isinstance(parameters, dict) else None + code = body.get("error_code") + description = body.get("description") + raise TelegramError( + f"Telegram {method} failed: {description if isinstance(description, str) else status}", + error_code=code if isinstance(code, int) else status, + retry_after=retry_after if isinstance(retry_after, int) and retry_after > 0 else None, + ) + + +class TelegramBotClient: + """One-attempt Bot API calls; messages never carry the token-bearing URL.""" + + def __init__(self, token: str, *, api_root: str = API_ROOT) -> None: + self._token = token + self._api_root = api_root.rstrip("/") + self._session: aiohttp.ClientSession | None = None + self._closed = False + + def _open_session(self) -> aiohttp.ClientSession: + if self._closed: + raise TelegramError("Telegram client is closed") + if self._session is None: + self._session = aiohttp.ClientSession() + return self._session + + async def _post(self, method: str, *, request_timeout: float, **request: Any) -> Any: + session = self._open_session() + try: + async with session.post( + f"{self._api_root}/bot{self._token}/{method}", + allow_redirects=False, + timeout=aiohttp.ClientTimeout(total=request_timeout), + **request, + ) as response: + status = response.status + try: + body = await response.json(content_type=None) + except ValueError: + body = None + except (aiohttp.ClientError, TimeoutError): + raise TelegramError( + f"Telegram {method} request failed; delivery may be uncertain" + ) from None + return _unwrap_result(method, status, body) + + async def call( + self, method: str, payload: dict | None = None, *, request_timeout: float = REQUEST_TIMEOUT, + ) -> Any: + return await self._post(method, request_timeout=request_timeout, json=payload or {}) + + async def upload( + self, *, method: str, fields: dict, file_field: str, filename: str, content: bytes, + ) -> Any: + form = aiohttp.FormData() + for key, value in fields.items(): + form.add_field(key, json.dumps(value) if isinstance(value, dict | list) else str(value)) + form.add_field( + file_field, content, filename=filename, content_type="application/octet-stream", + ) + return await self._post(method, request_timeout=UPLOAD_TIMEOUT, data=form) + + async def download_file(self, file_id: str, *, max_bytes: int) -> tuple[str, bytes]: + file = await self.call("getFile", {"file_id": file_id}) + file = file if isinstance(file, dict) else {} + size, path = file.get("file_size"), file.get("file_path") + if isinstance(size, int) and size > max_bytes: + raise _download_limit_error(max_bytes) + if ( + not isinstance(path, str) + or not path + or path.startswith("/") + or ".." in path.split("/") + ): + raise ValueError("This Telegram file is unavailable. Send the attachment again.") + session = self._open_session() + try: + async with session.get( + f"{self._api_root}/file/bot{self._token}/{quote(path)}", + allow_redirects=False, + timeout=aiohttp.ClientTimeout(total=DOWNLOAD_TIMEOUT), + ) as response: + if response.status != 200: + raise RuntimeError(f"Telegram file download failed (HTTP {response.status})") + chunks, received = [], 0 + async for chunk in response.content.iter_chunked(DOWNLOAD_CHUNK): + received += len(chunk) + if received > max_bytes: + raise _download_limit_error(max_bytes) + chunks.append(chunk) + except (aiohttp.ClientError, TimeoutError): + raise TelegramError("Telegram file download failed") from None + return path, b"".join(chunks) + + @property + def is_closed(self) -> bool: + return self._closed and (self._session is None or self._session.closed) + + async def close(self) -> None: + self._closed = True + if self._session: + await self._session.close() From ea709541e01cd2558eb151a8a1ece0ed751b5a8b Mon Sep 17 00:00:00 2001 From: CatalanCabbage Date: Wed, 14 Oct 2026 02:06:32 +0530 Subject: [PATCH 2/5] Test the Telegram adapter and make capability fallbacks capability-aware --- tests/channels/telegram/__init__.py | 0 tests/channels/telegram/test_events.py | 162 +++++++++++ tests/channels/telegram/test_outbound.py | 316 ++++++++++++++++++++++ tests/channels/telegram/test_polling.py | 271 +++++++++++++++++++ tests/channels/telegram/test_rendering.py | 269 ++++++++++++++++++ tests/channels/telegram/test_transport.py | 85 ++++++ tests/channels/telegram/test_webhook.py | 124 +++++++++ tests/contracts/test_imports.py | 5 + tests/contracts/test_providers.py | 41 ++- tests/support/telegram.py | 276 +++++++++++++++++++ 10 files changed, 1540 insertions(+), 9 deletions(-) create mode 100644 tests/channels/telegram/__init__.py create mode 100644 tests/channels/telegram/test_events.py create mode 100644 tests/channels/telegram/test_outbound.py create mode 100644 tests/channels/telegram/test_polling.py create mode 100644 tests/channels/telegram/test_rendering.py create mode 100644 tests/channels/telegram/test_transport.py create mode 100644 tests/channels/telegram/test_webhook.py create mode 100644 tests/support/telegram.py diff --git a/tests/channels/telegram/__init__.py b/tests/channels/telegram/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/channels/telegram/test_events.py b/tests/channels/telegram/test_events.py new file mode 100644 index 0000000..cf2721a --- /dev/null +++ b/tests/channels/telegram/test_events.py @@ -0,0 +1,162 @@ +import pytest + +from tests.support.telegram import ( + BOT, + callback_update, + group_update, + mention, + read, + telegram_message, + telegram_update, +) + + +def test_private_message_identity_and_metadata(): + message = read(telegram_update()) + assert message.id == "telegram:99:30:100" + assert message.conversation_id == "telegram:99:30" + assert message.channel == "telegram" and message.role == "user" + assert message.sender.id == "30" and message.sender.display_name == "Alice Liddell" + assert message.text == "hello" and message.thread_id is None + assert dict(message.metadata) == { + "bot_id": 99, "chat_id": 30, "chat_type": "private", + "message_id": 100, "message_thread_id": None, + } + + +@pytest.mark.parametrize("token", ["@agentchat_bot", "@AgentChat_Bot"]) +def test_group_mention_is_addressed_and_stripped(token): + text = f"{token} what is up" + message = read(group_update(text, entities=[mention(text, token)])) + assert message.text == "what is up" + assert message.conversation_id == "telegram:99:-1001" + + +def test_mention_after_astral_emoji_is_sliced_in_utf16_units(): + text = "๐Ÿ˜€ @agentchat_bot hi ๆ—ฅๆœฌ" + message = read(group_update(text)) + assert message.text == "๐Ÿ˜€ hi ๆ—ฅๆœฌ" + + +def test_command_with_bot_suffix_keeps_the_command(): + text = "/status@agentchat_bot now" + entity = mention(text, "/status@agentchat_bot", "bot_command") + message = read(group_update(text, entities=[entity])) + assert message.text == "/status now" + + +def test_text_mention_and_reply_to_bot_are_addressed(): + text = "Bot please" + entity = mention(text, "Bot", "text_mention", user={"id": 99, "is_bot": True}) + assert read(group_update(text, entities=[entity])).text == "Bot please" + reply = {"message_id": 90, "from": dict(BOT), "chat": {"id": -1001, "type": "supergroup"}} + assert read(group_update("thanks", reply_to_message=reply)).text == "thanks" + + +@pytest.mark.parametrize( + "update", + [ + group_update("just chatting"), + group_update("@other_bot hi", entities=[mention("@other_bot hi", "@other_bot")]), + group_update( + "/status@other_bot", + entities=[mention("/status@other_bot", "/status@other_bot", "bot_command")], + ), + group_update("hi", entities=[{"type": "text_mention", "offset": 0, "length": 2, + "user": {"id": 7}}]), + group_update("@agentchat_bot hi", entities=[{"type": "mention", "offset": 3, + "length": 99}]), + telegram_update(**{"from": {**BOT, "id": 77}}), + telegram_update(**{"from": {"id": 30}}), + telegram_update(chat={"id": -100, "type": "channel", "title": "News"}), + telegram_update(message_id=0), + telegram_update(text=None, new_chat_members=[{"id": 5, "is_bot": False}]), + telegram_update(text=" "), + {"update_id": 1}, + {"update_id": 1, "message": "hello"}, + {"update_id": 1, "edited_message": telegram_message()}, + None, + [], + ], +) +def test_unaddressed_bot_channel_service_and_malformed_updates_are_dropped(update): + assert read(update) is None + + +def test_forum_topic_is_a_thread_but_reply_threads_are_not(): + topic = read(telegram_update(message_thread_id=7, is_topic_message=True)) + assert topic.thread_id == "7" + assert topic.conversation_id == "telegram:99:30:7" + assert topic.metadata["message_thread_id"] == 7 + reply_thread = read(telegram_update(message_thread_id=7)) + assert reply_thread.thread_id is None + assert reply_thread.conversation_id == "telegram:99:30" + assert reply_thread.metadata["message_thread_id"] is None + + +def test_attachments_use_the_largest_photo_and_document_details(): + photo = read(telegram_update(text=None, caption="look", photo=[ + {"file_id": "small", "file_unique_id": "s", "width": 90, "height": 90}, + {"file_id": "large", "file_unique_id": "l", "width": 800, "height": 800, + "file_size": 5000}, + ])) + assert photo.text == "look" + assert [(a.id, a.name, a.media_type, a.size) for a in photo.attachments] == [ + ("large", None, "image/jpeg", 5000), + ] + document = {"file_id": "doc", "file_unique_id": "d", "file_name": "report.pdf", + "mime_type": "application/pdf", "file_size": 12} + message = read(telegram_update(text=None, document=document)) + assert message.text == "" + assert [(a.id, a.name, a.media_type, a.size) for a in message.attachments] == [ + ("doc", "report.pdf", "application/pdf", 12), + ] + gif = {"file_id": "gif", "file_unique_id": "g", "file_name": "a.mp4", "mime_type": "video/mp4"} + animated = read(telegram_update(text=None, animation=gif, document=gif)) + assert [a.id for a in animated.attachments] == ["gif"] + voice = read(telegram_update(text=None, voice={"file_id": "v", "mime_type": "audio/ogg"})) + assert voice.attachments[0].media_type == "audio/ogg" + sticker = read(telegram_update(text=None, sticker={"file_id": "st", "type": "regular"})) + assert sticker.attachments[0].name is None and sticker.attachments[0].media_type is None + + +def test_group_caption_mention_uses_caption_entities(): + caption = "@agentchat_bot what is this" + message = read(group_update( + text=None, entities=[], caption=caption, + caption_entities=[mention(caption, "@agentchat_bot")], + document={"file_id": "doc"}, + )) + assert message.text == "what is this" + assert message.attachments[0].id == "doc" + + +def test_callback_tap_becomes_the_button_label(): + message = read(callback_update()) + assert message.id == "telegram:99:callback:cb-1" + assert message.conversation_id == "telegram:99:30" + assert message.text == "Yes please" + assert message.sender.id == "30" + assert message.metadata["message_id"] == 55 + assert message.metadata["callback_query_id"] == "cb-1" + assert message.metadata["callback_data"] == "yes" + assert read(callback_update(data="unknown")).text == "unknown" + group = callback_update() + group["callback_query"]["message"]["chat"] = {"id": -1001, "type": "supergroup"} + assert read(group).conversation_id == "telegram:99:-1001" + + +@pytest.mark.parametrize( + "changes", + [ + {"data": None}, + {"from": {**BOT, "id": 77}}, + {"message": {"message_id": 55, "date": 0, "chat": {"id": 30, "type": "private"}}}, + {"message": None}, + {"id": ""}, + ], +) +def test_unusable_callbacks_are_dropped(changes): + update = callback_update() + update["callback_query"].update(changes) + assert read(update) is None diff --git a/tests/channels/telegram/test_outbound.py b/tests/channels/telegram/test_outbound.py new file mode 100644 index 0000000..f4c50d3 --- /dev/null +++ b/tests/channels/telegram/test_outbound.py @@ -0,0 +1,316 @@ +import asyncio +from dataclasses import replace + +import pytest + +from agentchat import RichReply, UploadFile +from agentchat.channels.telegram import Telegram +from agentchat.channels.telegram import adapter as telegram_adapter +from agentchat.channels.telegram.events import normalize +from agentchat.channels.telegram.transport import TelegramError +from tests.support.receipts import Receipt +from tests.support.telegram import ( + HOOK_URL, + SECRET, + TOKEN, + FakeBotAPI, + TelegramClient, + callback_update, + faked, + read, + served, + telegram_update, + until, +) + + +@pytest.mark.asyncio +async def test_reply_payload_routes_to_the_source_and_quotes_it(): + client = TelegramClient() + channel = faked(client) + source = read(telegram_update(message_thread_id=7, is_topic_message=True)) + try: + reply = await channel.reply(source, "**done**") + (_, payload), _ = client.receipt.calls[0] + assert payload == { + "chat_id": 30, + "text": "done", + "entities": [{"type": "bold", "offset": 0, "length": 4}], + "link_preview_options": {"is_disabled": True}, + "reply_parameters": {"message_id": 100, "allow_sending_without_reply": True}, + "message_thread_id": 7, + } + assert reply.id == "telegram:99:30:101" + assert reply.conversation_id == source.conversation_id + assert reply.thread_id == "7" + assert reply.text == "**done**" + assert reply.sender.id == "99" + assert reply.metadata["message_id"] == 101 + await channel.reply(read(telegram_update()), "plain") + assert "message_thread_id" not in client.receipt.calls[1][0][1] + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_reply_to_a_callback_quotes_the_keyboard_message(): + client = TelegramClient() + channel = faked(client) + try: + await channel.reply(read(callback_update()), "ok") + assert client.receipt.calls[0][0][1]["reply_parameters"]["message_id"] == 55 + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_reply_to_a_callback_drops_tap_only_metadata(): + client = TelegramClient() + channel = faked(client) + try: + reply = await channel.reply(read(callback_update()), "ok") + assert client.receipt.calls[0][0][1]["reply_parameters"]["message_id"] == 55 + assert "callback_query_id" not in reply.metadata + assert "callback_data" not in reply.metadata + assert reply.metadata["message_id"] == 101 + finally: + await channel.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "metadata", + [{"bot_id": 98}, {"chat_id": "30"}, {"message_id": True}, {"message_thread_id": "7"}], +) +async def test_reply_rejects_sources_from_another_bot_or_malformed_routes(metadata): + client = TelegramClient() + channel = faked(client) + source = read(telegram_update()) + source = replace(source, metadata={**source.metadata, **metadata}) + try: + with pytest.raises(ValueError, match="does not belong"): + await channel.reply(source, "answer") + assert not client.calls + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_reply_rich_passes_reply_markup_through(): + client = TelegramClient() + channel = faked(client) + source = read(telegram_update()) + keyboard = {"inline_keyboard": [[{"text": "Yes", "callback_data": "yes"}]]} + try: + reply = await channel.reply_rich(source, RichReply(text="Continue?", blocks=(keyboard,))) + payload = client.receipt.calls[0][0][1] + assert payload["reply_markup"] == keyboard + assert payload["text"] == "Continue?" + assert reply.text == "Continue?" + for content in ( + RichReply(text="x"), + RichReply(text="x", blocks=(keyboard, keyboard)), + RichReply(text="x", blocks=(keyboard,), attachments=({"a": 1},)), + ): + with pytest.raises(ValueError): + await channel.reply_rich(source, content) + assert len(client.receipt.calls) == 1 + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_typing_keepalive_refreshes_until_cleared(monkeypatch): + monkeypatch.setattr(telegram_adapter, "TYPING_INTERVAL", 0.01) + client = TelegramClient() + channel = faked(client) + source = read(telegram_update(message_thread_id=7, is_topic_message=True)) + try: + assert await channel.set_status(source, "thinking") is True + assert client.method_calls("sendChatAction")[0] == { + "chat_id": 30, "action": "typing", "message_thread_id": 7, + } + await until(lambda: len(client.method_calls("sendChatAction")) >= 3) + assert await channel.set_status(source, "") is True + assert not channel._typing + sent = len(client.method_calls("sendChatAction")) + await asyncio.sleep(0.05) + assert len(client.method_calls("sendChatAction")) == sent + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_typing_keepalive_expires_and_swallows_later_failures(monkeypatch): + monkeypatch.setattr(telegram_adapter, "TYPING_INTERVAL", 0.01) + monkeypatch.setattr(telegram_adapter, "TYPING_EXPIRY", 0.1) + client = TelegramClient() + channel = faked(client) + source = read(telegram_update()) + try: + await channel.set_status(source, "thinking") + task = channel._typing[(30, None)] + call = client.call + + async def failing(method, payload=None, **options): + if method == "sendChatAction": + client.calls.append((method, payload)) + raise TelegramError("Telegram sendChatAction failed: Too Many", error_code=429) + return await call(method, payload, **options) + + client.call = failing + await until(lambda: len(client.method_calls("sendChatAction")) >= 3) + await asyncio.wait_for(task, 1) + assert not channel._typing + with pytest.raises(TelegramError): + await channel.set_status(source, "thinking") + assert not channel._typing + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_close_cancels_typing_keepalive_before_closing_the_client(): + client = TelegramClient() + channel = faked(client) + await channel.set_status(read(telegram_update()), "thinking") + task = channel._typing[(30, None)] + await channel.close() + assert task.cancelled() and client.closed + with pytest.raises(RuntimeError, match="closed"): + await channel.set_status(read(telegram_update()), "thinking") + + +@pytest.mark.asyncio +async def test_upload_sends_documents_to_the_source_chat_and_thread(): + client = TelegramClient() + channel = faked(client) + source = read(telegram_update(message_thread_id=7, is_topic_message=True)) + try: + uploaded = await channel.upload_files(source, [ + UploadFile("a.txt", b"a", title="๐Ÿ˜€" * 600), + UploadFile("b.txt", b"b"), + ]) + assert [item.id for item in uploaded] == ["file-1", "file-2"] + method, fields, field, filename, content = client.uploads[0] + assert (method, field, filename, content) == ("sendDocument", "document", "a.txt", b"a") + assert fields["chat_id"] == 30 and fields["message_thread_id"] == 7 + assert fields["caption"] == "๐Ÿ˜€" * 512 + assert "caption" not in client.uploads[1][1] + for files in ([], [UploadFile("../a", b"a")], [UploadFile("a.txt", b"")]): + with pytest.raises(ValueError): + await channel.upload_files(source, files) + assert len(client.uploads) == 2 + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_upload_over_http_is_multipart_with_a_document_receipt(): + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + source = normalize(telegram_update(), bot_id=42, bot_username="fake_bot") + try: + uploaded = await channel.upload_files(source, [UploadFile("a.txt", b"hi", "Report")]) + assert uploaded[0].id == "uploaded-file" and uploaded[0].permalink is None + payload = api.method_calls("sendDocument")[0] + assert payload == {"chat_id": "30", "caption": "Report", "document": ("a.txt", b"hi")} + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_download_is_bounded_and_never_reveals_the_token(): + async with FakeBotAPI(TOKEN) as api: + api.files = { + "doc": {"info": {"file_id": "doc", "file_path": "documents/file_1.pdf"}, + "content": b"x" * 10}, + "photo": {"info": {"file_id": "photo", "file_path": "photos/file_2.jpg", + "file_size": 100}, "content": b"y" * 100}, + } + channel = served(api) + message = normalize(telegram_update( + text=None, document={"file_id": "doc", "file_name": "report.pdf"}, + ), bot_id=42, bot_username="fake_bot") + photo = normalize(telegram_update( + text=None, photo=[{"file_id": "photo"}], + ), bot_id=42, bot_username="fake_bot") + try: + downloaded = await channel.download_attachment(message, message.attachments[0]) + assert downloaded.content == b"x" * 10 and downloaded.name == "report.pdf" + downloaded = await channel.download_attachment(photo, photo.attachments[0]) + assert downloaded.name == "file_2.jpg" and downloaded.media_type == "image/jpeg" + with pytest.raises(ValueError, match="download limit"): + await channel.download_attachment(photo, photo.attachments[0], max_bytes=50) + with pytest.raises(ValueError, match="download limit"): + await channel.download_attachment(message, message.attachments[0], max_bytes=5) + assert len(api.method_calls("getFile")) == 4 + api.files["doc"]["content"] = None + with pytest.raises(RuntimeError, match="HTTP 404") as caught: + await channel.download_attachment(message, message.attachments[0]) + assert TOKEN not in str(caught.value) and TOKEN not in repr(caught.value) + api.files["doc"]["info"]["file_path"] = "../secret" + with pytest.raises(ValueError, match="unavailable"): + await channel.download_attachment(message, message.attachments[0]) + with pytest.raises(ValueError, match="does not belong"): + await channel.download_attachment(message, photo.attachments[0]) + with pytest.raises(ValueError, match="does not belong"): + await channel.download_attachment( + replace(message, role="assistant"), message.attachments[0], + ) + with pytest.raises(ValueError, match="byte limit"): + await channel.download_attachment(message, message.attachments[0], max_bytes=0) + finally: + await channel.close() + + +@pytest.mark.parametrize( + ("options", "error"), + [ + ({"bot_token": " "}, "required"), + ({"bot_token": "fake"}, "123456:ABC"), + ({"bot_token": "abc:def"}, "123456:ABC"), + ({"webhook_url": HOOK_URL}, "together"), + ({"webhook_secret": SECRET}, "together"), + ({"webhook_url": "http://example.test/webhooks/telegram", "webhook_secret": SECRET}, + "https"), + ({"webhook_url": "https://example.test/hook", "webhook_secret": SECRET}, "https"), + ({"webhook_url": HOOK_URL, "webhook_secret": "bad secret"}, "webhook_secret"), + ({"webhook_url": HOOK_URL, "webhook_secret": "x" * 257}, "webhook_secret"), + ], +) +def test_constructor_validation(options, error): + with pytest.raises(ValueError, match=error): + Telegram(**{"bot_token": "99:fake", **options}) + + +def test_from_env(monkeypatch): + for key in ("TELEGRAM_BOT_TOKEN", "TELEGRAM_WEBHOOK_URL", "TELEGRAM_WEBHOOK_SECRET"): + monkeypatch.delenv(key, raising=False) + with pytest.raises(RuntimeError, match="TELEGRAM_BOT_TOKEN is required"): + Telegram.from_env() + monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "99:fake") + assert Telegram.from_env()._webhook_url is None + with pytest.raises(RuntimeError, match="webhook_url and webhook_secret"): + Telegram.from_env().create_app() + monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", HOOK_URL) + with pytest.raises(RuntimeError, match="required together"): + Telegram.from_env() + monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", SECRET) + monkeypatch.setenv("TELEGRAM_PORT", "4000") + channel = Telegram.from_env() + assert (channel._webhook_url, channel._port, channel._bot_id) == (HOOK_URL, 4000, 99) + + +@pytest.mark.asyncio +async def test_reply_before_run_and_receipt_failures(): + receipt = Receipt(None) + channel = faked(TelegramClient(receipt)) + try: + with pytest.raises(RuntimeError, match="receipt"): + await channel.reply(read(telegram_update()), "answer") + receipt.value = True + with pytest.raises(RuntimeError, match="receipt"): + await channel.reply(read(telegram_update()), "answer") + finally: + await channel.close() diff --git a/tests/channels/telegram/test_polling.py b/tests/channels/telegram/test_polling.py new file mode 100644 index 0000000..ab85f03 --- /dev/null +++ b/tests/channels/telegram/test_polling.py @@ -0,0 +1,271 @@ +import asyncio +import time + +import pytest + +from agentchat import AgentChat +from agentchat.channels.telegram.adapter import retry_delay +from agentchat.channels.telegram.transport import TelegramError +from tests.support.telegram import ( + TOKEN, + FakeBotAPI, + TelegramClient, + callback_update, + faked, + fast_retries, + group_update, + nothing, + served, + telegram_update, + until, +) + + +@pytest.mark.asyncio +async def test_callback_tap_is_delivered_and_answered(): + client = TelegramClient(batches=[[callback_update(update_id=5)]]) + channel = faked(client) + seen = [] + + async def receive(channel, message): + seen.append(message.text) + + channel.bind(receive) + running = asyncio.create_task(channel.run()) + try: + await until(lambda: client.method_calls("answerCallbackQuery") and client.started.is_set()) + assert seen == ["Yes please"] + assert client.method_calls("answerCallbackQuery") == [{"callback_query_id": "cb-1"}] + assert client.method_calls("getUpdates")[1]["offset"] == 6 + finally: + await channel.close() + await running + + +@pytest.mark.asyncio +async def test_close_cancels_inflight_long_poll_promptly(): + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + channel.bind(nothing) + running = asyncio.create_task(channel.run()) + await asyncio.wait_for(api.poll_started.wait(), 2) + started = time.monotonic() + await channel.close() + assert time.monotonic() - started < 0.5 + assert await asyncio.wait_for(running, 0.5) is None + assert api.method_calls("deleteWebhook") == [{"drop_pending_updates": False}] + poll = api.method_calls("getUpdates")[0] + assert poll == {"timeout": 30, "allowed_updates": ["message", "callback_query"]} + assert channel._client.is_closed + + +@pytest.mark.asyncio +async def test_real_poll_to_reply_round_trip_through_agentchat(): + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + app = AgentChat(channels=[channel]) + + @app.on_message + async def respond(context): + return f"**echo** {context.message.text}" + + await api.push(telegram_update(text="hi")) + running = asyncio.create_task(app.run()) + try: + await until(lambda: api.method_calls("sendMessage")) + sent = api.method_calls("sendMessage")[0] + assert sent["text"] == "echo hi" + assert sent["entities"] == [{"type": "bold", "offset": 0, "length": 4}] + assert sent["reply_parameters"] == {"message_id": 100, + "allow_sending_without_reply": True} + history = await app.state.history("telegram:42:30") + assert history[-1].id == "telegram:42:30:501" + assert history[-1].sender.display_name == "Bot" + finally: + await app.close() + await running + + +@pytest.mark.asyncio +async def test_full_queue_holds_offset_until_update_is_accepted(monkeypatch): + fast_retries(monkeypatch) + release = asyncio.Event() + seen = [] + + async def handler(channel, message): + seen.append(message.text) + if len(seen) == 1: + await release.wait() + + async with FakeBotAPI(TOKEN) as api: + channel = served(api, max_pending_events=1) + channel.bind(handler) + await api.push(telegram_update(text="one", message_id=1)) + second = await api.push(telegram_update(text="two", message_id=2)) + running = asyncio.create_task(channel.run()) + await until(lambda: seen == ["one"]) + await until(lambda: api.offsets.count(second) >= 2) + assert all(o is None or o <= second for o in api.offsets) + assert seen == ["one"] + release.set() + await until(lambda: seen == ["one", "two"]) + await until(lambda: second + 1 in api.offsets) + await channel.close() + await running + offsets = [o for o in api.offsets if o is not None] + assert offsets == sorted(offsets) + + +@pytest.mark.asyncio +async def test_unaddressed_updates_advance_the_offset(): + seen = [] + + async def handler(channel, message): + seen.append(message.text) + + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + channel.bind(handler) + skipped = await api.push(group_update("chatter", message_id=1)) + await api.push(telegram_update(text="direct", message_id=2)) + running = asyncio.create_task(channel.run()) + await until(lambda: seen == ["direct"]) + await until(lambda: skipped + 2 in api.offsets) + await channel.close() + await running + + +@pytest.mark.asyncio +async def test_close_confirms_accepted_offset(): + seen = [] + + async def handler(channel, message): + seen.append(message.text) + + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + channel.bind(handler) + update = await api.push(telegram_update(text="one")) + running = asyncio.create_task(channel.run()) + await until(lambda: seen == ["one"]) + await channel.close() + await running + assert api.method_calls("getUpdates")[-1] == { + "offset": update + 1, "limit": 1, "timeout": 0, + } + assert api.updates == [] + + +@pytest.mark.asyncio +async def test_conflict_ends_run_and_closes_session(): + async with FakeBotAPI(TOKEN) as api: + api.failures["getUpdates"] = [ + (409, {"ok": False, "error_code": 409, "description": "Conflict"}), + ] + channel = served(api) + channel.bind(nothing) + with pytest.raises(TelegramError, match="Conflict") as caught: + await asyncio.wait_for(channel.run(), 2) + assert caught.value.error_code == 409 + assert channel._client.is_closed + assert channel._delivery.closed + + +@pytest.mark.asyncio +@pytest.mark.parametrize("code", [401, 404]) +async def test_invalid_token_is_fatal(code): + async with FakeBotAPI(TOKEN) as api: + api.failures["getMe"] = [(code, {"ok": False, "error_code": code, "description": "Bad"})] + channel = served(api) + channel.bind(nothing) + with pytest.raises(TelegramError, match="authentication failed") as caught: + await asyncio.wait_for(channel.run(), 2) + assert caught.value.error_code == code + assert channel._client.is_closed + + +@pytest.mark.asyncio +async def test_token_for_another_bot_is_rejected(): + async with FakeBotAPI(TOKEN) as api: + api.bot["id"] = 7 + channel = served(api) + channel.bind(nothing) + with pytest.raises(RuntimeError, match="does not match"): + await asyncio.wait_for(channel.run(), 2) + assert not api.method_calls("getUpdates") + + +@pytest.mark.asyncio +async def test_server_error_backs_off_then_continues(monkeypatch, caplog): + fast_retries(monkeypatch) + seen = [] + + async def handler(channel, message): + seen.append(message.text) + + async with FakeBotAPI(TOKEN) as api: + api.failures["getUpdates"] = [ + (500, {"ok": False, "error_code": 500, "description": "Internal"}), + (502, None), + (429, {"ok": False, "error_code": 429, "description": "Too Many", + "parameters": {"retry_after": 1}}), + ] + channel = served(api) + channel.bind(handler) + await api.push(telegram_update(text="after errors")) + running = asyncio.create_task(channel.run()) + started = time.monotonic() + await until(lambda: seen == ["after errors"]) + assert time.monotonic() - started >= 1 + await channel.close() + await running + assert len([r for r in caplog.records if "retrying" in r.message]) == 3 + + +def test_retry_delay_schedule(): + assert [retry_delay(n, None) for n in range(7)] == [1, 2, 4, 8, 16, 30, 30] + assert retry_delay(0, 45) == 45 + + +@pytest.mark.asyncio +async def test_external_cancel_of_run_cleans_up(): + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + channel.bind(nothing) + running = asyncio.create_task(channel.run()) + await asyncio.wait_for(api.poll_started.wait(), 2) + running.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(running, 0.5) + assert channel._worker.done() + assert channel._client.is_closed + assert channel._delivery.closed + + +@pytest.mark.asyncio +async def test_close_from_inside_handler(): + finished = asyncio.Event() + + async def handler(channel, message): + await channel.close() + finished.set() + + async with FakeBotAPI(TOKEN) as api: + channel = served(api) + channel.bind(handler) + await api.push(telegram_update(text="stop")) + running = asyncio.create_task(channel.run()) + await asyncio.wait_for(finished.wait(), 2) + assert await asyncio.wait_for(running, 1) is None + assert channel._client.is_closed + await until(lambda: not channel._delivery.tasks) + + +@pytest.mark.asyncio +async def test_close_before_run_and_double_run(): + channel = faked() + channel.bind(nothing) + await channel.close() + await channel.close() + with pytest.raises(RuntimeError, match="cannot be restarted"): + await channel.run() diff --git a/tests/channels/telegram/test_rendering.py b/tests/channels/telegram/test_rendering.py new file mode 100644 index 0000000..63b6494 --- /dev/null +++ b/tests/channels/telegram/test_rendering.py @@ -0,0 +1,269 @@ +import random + +import pytest + +from agentchat.channels.telegram.rendering import render, text_payload + + +def _units(text): + return text.encode("utf-16-le") + + +def _slice(text, entity): + start = entity["offset"] * 2 + return _units(text)[start:start + entity["length"] * 2].decode("utf-16-le") + + +def _only(entities, entity_type): + found = [e for e in entities if e["type"] == entity_type] + assert len(found) == 1, entities + return found[0] + + +def _types(markdown): + return [e["type"] for e in render(markdown)[1]] + + +def test_rendering_snake_case_identifiers_are_not_italicised(): + text, entities = render("call get_user_data() then set_config_value()") + assert text == "call get_user_data() then set_config_value()" + assert entities == [] + + +def test_rendering_fenced_block_becomes_pre_with_language(): + text, entities = render("intro\n\n```ts\nconst a = 1;\n```") + pre = _only(entities, "pre") + assert pre["language"] == "ts" + assert _slice(text, pre) == "const a = 1;" + + +def test_rendering_unclosed_fence_still_yields_a_pre_block(): + text, entities = render("```js\nconst a = 1;\nconst b = 2;") + assert _slice(text, _only(entities, "pre")) == "const a = 1;\nconst b = 2;" + + +def test_rendering_backticks_inside_a_fenced_block_survive(): + text, entities = render("```sh\necho `date`\n```") + assert text == "echo `date`" + assert [e["type"] for e in entities] == ["pre"] + + +def test_rendering_nested_emphasis_produces_both_entities(): + assert set(_types("**bold with `code` inside**")) == {"bold", "code"} + + +def test_rendering_offsets_are_utf16_code_units(): + text, entities = render("๐ŸŽ‰ `x` done") + code = _only(entities, "code") + assert code["offset"] == 3 + assert _slice(text, code) == "x" + + +def test_rendering_cjk_offsets_line_up(): + text, entities = render("ๆ—ฅๆœฌ่ชžใฎ `foo` ใƒ†ใ‚นใƒˆ") + assert _slice(text, _only(entities, "code")) == "foo" + + +def test_rendering_links_render_as_text_link(): + text, entities = render("see [docs](https://example.com) here") + link = _only(entities, "text_link") + assert link["url"] == "https://example.com" + assert _slice(text, link) == "docs" + + +def test_rendering_every_entity_stays_inside_the_text(): + text, entities = render( + "# Head\n\n- **a** and `b`\n- [l](http://x)\n\n```py\nx=1\n```\n\n> quoted" + ) + assert entities + for e in entities: + assert e["offset"] >= 0 + assert e["length"] > 0 + assert e["offset"] + e["length"] <= len(_units(text)) // 2 + + +def test_rendering_emoji_before_and_inside_entities(): + text, entities = render("๐Ÿ˜€ **bold ๐Ÿ˜€ x** and `ๆ—ฅๆœฌ๐Ÿ˜€`") + bold, code = _only(entities, "bold"), _only(entities, "code") + assert (bold["offset"], bold["length"]) == (3, 9) + assert _slice(text, bold) == "bold ๐Ÿ˜€ x" + assert _slice(text, code) == "ๆ—ฅๆœฌ๐Ÿ˜€" + assert code["length"] == 4 + + +def test_rendering_astral_emoji_then_bold(): + text, entities = render("๐Ÿ˜€ **bold**") + assert text == "๐Ÿ˜€ bold" + assert entities == [{"type": "bold", "offset": 3, "length": 4}] + + +@pytest.mark.parametrize("literal", ["edit __init__.py", "run __main__", "a __b__ c"]) +def test_rendering_double_underscore_is_not_a_delimiter(literal): + assert render(literal) == (literal, []) + + +@pytest.mark.parametrize( + "literal", + ["snake_case_name", "a*b*c", "get_user_data()", "x * y * z", "file_name_v2.py", "2*3*4"], +) +def test_rendering_intraword_and_spaced_markers_stay_literal(literal): + assert render(literal) == (literal, []) + + +def test_rendering_unclosed_fence_runs_to_end(): + text, entities = render("before\n\n~~~python\nx = 1\n\n**not bold**\n") + pre = _only(entities, "pre") + assert pre["language"] == "python" + assert _slice(text, pre) == "x = 1\n\n**not bold**" + assert text.startswith("before\n\n") + + +def test_rendering_fence_language_is_first_word_of_info_string(): + _, entities = render('```python title="x.py"\npass\n```') + assert _only(entities, "pre")["language"] == "python" + + +def test_rendering_fence_without_language_has_no_language_key(): + _, entities = render("```\npass\n```") + assert entities == [{"type": "pre", "offset": 0, "length": 4}] + + +def test_rendering_nested_bold_and_italic(): + text, entities = render("**bold _it_ end** and _**both**_") + assert text == "bold it end and both" + spans = [(e["type"], _slice(text, e)) for e in entities] + assert spans == [ + ("bold", "bold it end"), + ("italic", "it"), + ("italic", "both"), + ("bold", "both"), + ] + + +def test_rendering_link_containing_bold(): + text, entities = render("[**big** docs](https://x.y/z)") + assert text == "big docs" + assert entities == [ + {"type": "text_link", "offset": 0, "length": 8, "url": "https://x.y/z"}, + {"type": "bold", "offset": 0, "length": 3}, + ] + + +def test_rendering_code_inside_link_drops_code_entity(): + text, entities = render("[`run()`](https://x.y)") + assert text == "run()" + assert [e["type"] for e in entities] == ["text_link"] + + +def test_rendering_relative_link_stays_literal(): + assert render("[docs](./readme.md)") == ("[docs](./readme.md)", []) + + +def test_rendering_code_containing_markers_stays_literal(): + text, entities = render("`**not bold** _x_ [a](https://b.c)`") + assert text == "**not bold** _x_ [a](https://b.c)" + assert [e["type"] for e in entities] == ["code"] + + +def test_rendering_crlf_input(): + text, entities = render("# Title\r\n\r\n- one\r\n- **two**\r\n\r\n```sh\r\nls\r\n```\r\n") + assert "\r" not in text + assert text == "Title\n\nโ€ข one\nโ€ข two\n\nls" + assert [(e["type"], _slice(text, e)) for e in entities] == [ + ("bold", "Title"), + ("bold", "two"), + ("pre", "ls"), + ] + + +def test_rendering_blocks_and_lists(): + text, entities = render("1. first\n2. second\n * nested\n\n~~gone~~ ||secret||") + assert text == "1. first\n2. second\n โ€ข nested\n\ngone secret" + assert [(e["type"], _slice(text, e)) for e in entities] == [ + ("strikethrough", "gone"), + ("spoiler", "secret"), + ] + + +def test_rendering_consecutive_quote_lines_form_one_blockquote(): + text, entities = render("> one\n> **two**\nafter") + assert text == "one\ntwo\nafter" + assert [(e["type"], _slice(text, e)) for e in entities] == [ + ("blockquote", "one\ntwo"), + ("bold", "two"), + ] + + +def test_rendering_backslash_escapes_punctuation_only(): + assert render(r"\*not italic\* and \_x\_") == ("*not italic* and _x_", []) + assert render(r"C:\Users\name") == (r"C:\Users\name", []) + + +def test_text_payload_shape(): + assert text_payload("hi **you**") == { + "text": "hi you", + "entities": [{"type": "bold", "offset": 3, "length": 3}], + "link_preview_options": {"is_disabled": True}, + } + assert text_payload("plain") == { + "text": "plain", + "link_preview_options": {"is_disabled": True}, + } + + +@pytest.mark.parametrize("content", ["", " \n\t", "```\n```", "```\n\n\n```", None]) +def test_text_payload_rejects_empty(content): + with pytest.raises(ValueError, match="Telegram replies require nonempty text"): + text_payload(content) + + +def test_text_payload_limit_counts_utf16_units(): + assert text_payload("๐Ÿ˜€" * 2048)["text"] == "๐Ÿ˜€" * 2048 + with pytest.raises(ValueError, match="must fit 4,096 characters"): + text_payload("๐Ÿ˜€" * 2049) + + +def test_text_payload_limit_applies_to_rendered_text(): + assert text_payload("**" + "x" * 4096 + "**")["entities"][0]["length"] == 4096 + with pytest.raises(ValueError, match="must fit 4,096 characters"): + text_payload("x" * 4097) + + +_TOKENS = [ + "*", "**", "_", "__", "__init__.py", "~~", "||", "`", "```", "~~~", "\\", "\n", "\n\n", "\r\n", + "# ", "- ", "1. ", "> ", "[", "](", ")", "](https://x.y)", "https://a.b", "py", + "๐Ÿ˜€", "๐Ÿ‘จโ€๐Ÿ‘ฉโ€๐Ÿ‘ง", "ๆ—ฅๆœฌ่ชž", "รฉ", "a", "word", "b c", " ", "\t", "get_user_data()", +] + + +def _is_code_point_boundary(units, index): + if index * 2 >= len(units): + return True + unit = int.from_bytes(units[index * 2:index * 2 + 2], "little") + return not 0xDC00 <= unit <= 0xDFFF + + +def test_rendering_random_markdown_never_breaks_entity_bounds(): + rng = random.Random(4096) + for _ in range(2000): + source = "".join(rng.choice(_TOKENS) for _ in range(rng.randint(0, 60))) + text, entities = render(source) + units = _units(text) + total = len(units) // 2 + for e in entities: + assert e["offset"] >= 0, (source, e) + assert e["length"] > 0, (source, e) + assert e["offset"] + e["length"] <= total, (source, e) + assert _is_code_point_boundary(units, e["offset"]), (source, e) + assert _is_code_point_boundary(units, e["offset"] + e["length"]), (source, e) + assert entities == sorted(entities, key=lambda e: e["offset"]) + for n, outer in enumerate(entities): + outer_end = outer["offset"] + outer["length"] + for inner in entities[n + 1:]: + if inner["offset"] >= outer_end: + continue + assert inner["offset"] + inner["length"] <= outer_end, (source, outer, inner) + assert outer["type"] not in ("code", "pre"), (source, outer, inner) + assert not (outer["type"] == inner["type"] == "blockquote"), source + if outer["type"] == "text_link": + assert inner["type"] not in ("code", "pre", "text_link"), source diff --git a/tests/channels/telegram/test_transport.py b/tests/channels/telegram/test_transport.py new file mode 100644 index 0000000..5ffae32 --- /dev/null +++ b/tests/channels/telegram/test_transport.py @@ -0,0 +1,85 @@ +import asyncio +import logging +import socket + +import pytest + +from agentchat.channels.telegram import transport +from agentchat.channels.telegram.transport import TelegramBotClient, TelegramError +from tests.support.telegram import ( + TOKEN, + FakeBotAPI, + fast_retries, + nothing, + served, + until, +) + + +@pytest.mark.asyncio +async def test_long_poll_outlives_the_default_request_timeout(monkeypatch): + monkeypatch.setattr(transport, "REQUEST_TIMEOUT", 0.2) + async with FakeBotAPI(TOKEN) as api: + client = TelegramBotClient(TOKEN, api_root=api.root) + try: + assert await client.call("getUpdates", {"timeout": 1}, request_timeout=3) == [] + with pytest.raises(TelegramError, match="uncertain"): + await client.call("getUpdates", {"timeout": 1}, request_timeout=0.2) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_no_error_message_contains_token(monkeypatch, caplog): + fast_retries(monkeypatch) + caplog.set_level(logging.DEBUG) + errors = [] + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + dead_root = f"http://127.0.0.1:{sock.getsockname()[1]}" + async with FakeBotAPI(TOKEN) as api: + api.failures["getMe"] = [ + (401, {"ok": False, "error_code": 401, "description": "Unauthorized"}), + ] + api.failures["setWebhook"] = [(502, None)] + for root, method, request_timeout in [ + (dead_root, "getMe", 30), + (api.root, "getMe", 30), + (api.root, "setWebhook", 30), + (api.root, "getUpdates", 0.05), + ("not a url", "getMe", 30), + ]: + client = TelegramBotClient(TOKEN, api_root=root) + try: + payload = {"timeout": 1} if method == "getUpdates" else {} + with pytest.raises(TelegramError) as caught: + await client.call(method, payload, request_timeout=request_timeout) + errors.append(caught.value) + finally: + await client.close() + client = TelegramBotClient(TOKEN, api_root=dead_root) + + async def located(method, payload=None, **options): + return {"file_path": "documents/file.pdf"} + + client.call = located + try: + with pytest.raises(TelegramError) as caught: + await client.download_file("doc", max_bytes=10) + errors.append(caught.value) + finally: + await client.close() + api.failures["getUpdates"] = [(500, {"ok": False, "error_code": 500, "description": "x"})] + channel = served(api) + channel.bind(nothing) + running = asyncio.create_task(channel.run()) + await until(lambda: any("retrying" in r.message for r in caplog.records)) + await channel.close() + await running + for error in errors: + assert error.__cause__ is None + assert error.__context__ is None or error.__suppress_context__ + assert TOKEN not in str(error) and TOKEN not in repr(error) + # The fake server's own access log prints its request path; only our loggers matter. + ours = "\n".join(r.getMessage() for r in caplog.records if r.name != "aiohttp.access") + assert TOKEN not in ours and TOKEN.split(":")[1] not in ours diff --git a/tests/channels/telegram/test_webhook.py b/tests/channels/telegram/test_webhook.py new file mode 100644 index 0000000..e6e2173 --- /dev/null +++ b/tests/channels/telegram/test_webhook.py @@ -0,0 +1,124 @@ +import asyncio +import json +import socket + +import aiohttp +import pytest +from aiohttp.test_utils import TestClient, TestServer + +from agentchat.channels.telegram import Telegram +from agentchat.channels.telegram.adapter import SECRET_HEADER, WEBHOOK_PATH +from agentchat.channels.telegram.transport import TelegramBotClient +from tests.support.telegram import ( + GOOD, + HOOK_URL, + SECRET, + TOKEN, + FakeBotAPI, + TelegramClient, + body, + callback_update, + group_update, + hooked, + nothing, + until, +) + + +@pytest.mark.asyncio +async def test_webhook_statuses(): + client = TelegramClient() + channel = hooked(client, max_pending_events=1) + release = asyncio.Event() + + async def handler(channel, message): + await release.wait() + + channel.bind(handler) + try: + async with TestClient(TestServer(channel.create_app())) as http: + assert (await http.post(WEBHOOK_PATH, data=body(1))).status == 401 + wrong = {**GOOD, SECRET_HEADER: "nope"} + assert (await http.post(WEBHOOK_PATH, data=body(1), headers=wrong)).status == 401 + assert (await http.post(WEBHOOK_PATH, data=b"{not json", headers=GOOD)).status == 400 + assert (await http.post(WEBHOOK_PATH, data=b"[1]", headers=GOOD)).status == 400 + assert (await http.post(WEBHOOK_PATH, data=body(1), headers=GOOD)).status == 200 + assert client.method_calls("getMe") == [None] + unaddressed = json.dumps(group_update("chatter")) + assert (await http.post(WEBHOOK_PATH, data=unaddressed, headers=GOOD)).status == 200 + assert (await http.post(WEBHOOK_PATH, data=body(2), headers=GOOD)).status == 503 + release.set() + await until(lambda: not channel._delivery.tasks) + assert (await http.post(WEBHOOK_PATH, data=body(2), headers=GOOD)).status == 200 + await channel.close() + assert (await http.post(WEBHOOK_PATH, data=body(3), headers=GOOD)).status == 503 + assert client.method_calls("getMe") == [None] + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_webhook_returns_503_until_the_bot_identity_is_known(): + async with FakeBotAPI(TOKEN) as api: + api.failures["getMe"] = [(502, None)] + client = TelegramBotClient(TOKEN, api_root=api.root) + channel = Telegram(bot_token=TOKEN, client=client, webhook_url=HOOK_URL, + webhook_secret=SECRET) + channel.bind(nothing) + try: + async with TestClient(TestServer(channel.create_app())) as http: + assert (await http.post(WEBHOOK_PATH, data=body(1), headers=GOOD)).status == 503 + assert (await http.post(WEBHOOK_PATH, data=body(1), headers=GOOD)).status == 200 + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_webhook_callback_tap_is_answered(): + client = TelegramClient() + channel = hooked(client) + seen = [] + + async def handler(channel, message): + seen.append(message.text) + + channel.bind(handler) + try: + async with TestClient(TestServer(channel.create_app())) as http: + update = json.dumps(callback_update()) + assert (await http.post(WEBHOOK_PATH, data=update, headers=GOOD)).status == 200 + await until(lambda: client.method_calls("answerCallbackQuery")) + assert seen == ["Yes please"] + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_webhook_run_listens_before_setting_the_hook_and_leaves_it_set(): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + seen = [] + + async def handler(channel, message): + seen.append(message.text) + + async with FakeBotAPI(TOKEN) as api: + client = TelegramBotClient(TOKEN, api_root=api.root) + channel = Telegram(bot_token=TOKEN, client=client, webhook_url=HOOK_URL, + webhook_secret=SECRET, port=port) + channel.bind(handler) + running = asyncio.create_task(channel.run()) + await until(lambda: api.webhook is not None) + assert api.webhook == {"url": HOOK_URL, "secret_token": SECRET, + "allowed_updates": ["message", "callback_query"]} + async with aiohttp.ClientSession() as session: + async with session.post(f"http://127.0.0.1:{port}{WEBHOOK_PATH}", + data=body(5), headers=GOOD) as response: + assert response.status == 200 + await until(lambda: seen == ["hello"]) + await channel.close() + assert await asyncio.wait_for(running, 1) is None + assert channel._client.is_closed + assert api.webhook is not None + assert not api.method_calls("deleteWebhook") diff --git a/tests/contracts/test_imports.py b/tests/contracts/test_imports.py index 3acdfdb..af865b7 100644 --- a/tests/contracts/test_imports.py +++ b/tests/contracts/test_imports.py @@ -35,6 +35,8 @@ def test_core_and_namespaces_do_not_import_provider_or_framework_sdks(): import agentchat.channels.teams.events import agentchat.channels.imessage.events import agentchat.channels.whatsapp.events +import agentchat.channels.telegram.events +from agentchat.channels.telegram.rendering import text_payload assert 'agentchat.channels.slack' not in sys.modules assert 'agentchat.integrations.openai_agents' not in sys.modules """ @@ -85,6 +87,7 @@ def test_export_identity_in_both_import_orders(order): from agentchat.channels.teams.adapter import Teams as TeamsDirect from agentchat.channels.imessage.adapter import IMessage as IMessageDirect from agentchat.channels.whatsapp.adapter import WhatsApp as WhatsAppDirect +from agentchat.channels.telegram.adapter import Telegram as TelegramDirect from agentchat.integrations.openai_agents import should_reply as should_reply_direct """ namespace = """ @@ -93,12 +96,14 @@ def test_export_identity_in_both_import_orders(order): from agentchat.channels.teams import Teams from agentchat.channels.imessage import IMessage from agentchat.channels.whatsapp import WhatsApp +from agentchat.channels.telegram import Telegram from agentchat.integrations import should_reply """ checks = """ assert Slack is SlackDirect and SlackUser is SlackUserDirect assert Discord is DiscordDirect and Teams is TeamsDirect assert IMessage is IMessageDirect and WhatsApp is WhatsAppDirect +assert Telegram is TelegramDirect assert should_reply is should_reply_direct """ code = explicit + namespace if order == "explicit_first" else namespace + explicit diff --git a/tests/contracts/test_providers.py b/tests/contracts/test_providers.py index cdd0baa..bdc3667 100644 --- a/tests/contracts/test_providers.py +++ b/tests/contracts/test_providers.py @@ -6,25 +6,35 @@ from aiohttp import web from agentchat import AgentChat, RichReply, UploadFile +from agentchat.channels.base import ( + FileUploadChannel, + MirroringChannel, + RichReplyChannel, + StatusChannel, +) from agentchat.channels.discord import Discord from agentchat.channels.discord.events import normalize as discord_message from agentchat.channels.imessage import IMessage from agentchat.channels.imessage.events import normalize as imessage_message from agentchat.channels.teams import Teams from agentchat.channels.teams.events import normalize as teams_message +from agentchat.channels.telegram import Telegram +from agentchat.channels.telegram.events import normalize as telegram_message from agentchat.channels.whatsapp import WhatsApp from agentchat.channels.whatsapp.events import normalize as whatsapp_messages from tests.support.discord import DiscordClient, discord_event from tests.support.imessage import BubblesClient, SocketClient, imessage_event from tests.support.receipts import Receipt from tests.support.teams import APP_ID, TENANT_ID, teams_adapter, teams_event +from tests.support.telegram import TelegramClient, telegram_update from tests.support.whatsapp import WhatsAppSession, whatsapp_event -@pytest.fixture(params=["discord", "teams", "imessage", "whatsapp"]) +@pytest.fixture(params=["discord", "teams", "imessage", "whatsapp", "telegram"]) def harness(request): - receipt = Receipt(101 if request.param == "discord" else "sent") + receipt = Receipt(101 if request.param in ("discord", "telegram") else "sent") options = {"max_pending_events": 2, "shutdown_timeout": 0.01} + capabilities = set() if request.param == "discord": client = DiscordClient(receipt) channel = Discord(bot_token="fake", client=client, send_client=client, **options) @@ -50,6 +60,10 @@ def harness(request): **options, ) source = imessage_message(imessage_event(), instance_id="test") + elif request.param == "telegram": + channel = Telegram(bot_token="99:fake", client=TelegramClient(receipt), **options) + source = telegram_message(telegram_update(), bot_id=99, bot_username="agentchat_bot") + capabilities = {StatusChannel, RichReplyChannel, FileUploadChannel} else: channel = WhatsApp( phone_number_id="123", @@ -61,7 +75,9 @@ def harness(request): ) channel._session = WhatsAppSession(receipt) source = whatsapp_messages(whatsapp_event(), phone_number_id="123")[0] - return SimpleNamespace(channel=channel, source=source, receipt=receipt) + return SimpleNamespace( + channel=channel, source=source, receipt=receipt, capabilities=capabilities, + ) @pytest.mark.asyncio @@ -124,12 +140,19 @@ async def test_rejects_wrong_provider_and_empty_text_before_send(harness): @pytest.mark.asyncio async def test_optional_capability_fallbacks(harness): app = AgentChat(channels=[harness.channel]) + for protocol in (StatusChannel, RichReplyChannel, FileUploadChannel, MirroringChannel): + assert isinstance(harness.channel, protocol) == (protocol in harness.capabilities) try: - assert await app.set_status(harness.channel, harness.source, "working") is False - with pytest.raises(TypeError, match="rich replies"): - await app.reply_rich(harness.channel, harness.source, RichReply(text="answer")) - with pytest.raises(TypeError, match="file uploads"): - await app.upload_files(harness.channel, harness.source, [UploadFile("a.txt", b"a")]) + if StatusChannel not in harness.capabilities: + assert await app.set_status(harness.channel, harness.source, "working") is False + if RichReplyChannel not in harness.capabilities: + with pytest.raises(TypeError, match="rich replies"): + await app.reply_rich(harness.channel, harness.source, RichReply(text="answer")) + if FileUploadChannel not in harness.capabilities: + with pytest.raises(TypeError, match="file uploads"): + await app.upload_files( + harness.channel, harness.source, [UploadFile("a.txt", b"a")], + ) with pytest.raises(TypeError, match="mirroring"): await app.mirror(harness.channel, harness.source, harness.source) finally: @@ -201,7 +224,7 @@ async def start(site): running = asyncio.create_task(harness.channel.run()) try: async with asyncio.timeout(2): - if harness.channel.name == "discord": + if harness.channel.name in ("discord", "telegram"): await harness.channel._client.started.wait() elif harness.channel.name == "imessage": await harness.channel._socket.started.wait() diff --git a/tests/support/telegram.py b/tests/support/telegram.py new file mode 100644 index 0000000..99a849d --- /dev/null +++ b/tests/support/telegram.py @@ -0,0 +1,276 @@ +import asyncio +import json + +from aiohttp import web +from aiohttp.test_utils import TestServer + +from agentchat.channels.telegram import Telegram +from agentchat.channels.telegram import adapter as telegram_adapter +from agentchat.channels.telegram.adapter import SECRET_HEADER +from agentchat.channels.telegram.events import normalize +from agentchat.channels.telegram.transport import TelegramBotClient +from tests.support.receipts import Receipt + +TOKEN = "42:SECRET-token_abcdef" +SECRET = "hook_secret-1" +HOOK_URL = "https://example.test/webhooks/telegram" +GOOD = {SECRET_HEADER: SECRET, "Content-Type": "application/json"} +BOT = {"id": 99, "is_bot": True, "first_name": "Bot", "username": "agentchat_bot"} +ALICE = {"id": 30, "is_bot": False, "first_name": "Alice", "last_name": "Liddell"} +GROUP = {"id": -1001, "type": "supergroup", "title": "Team"} + + +def telegram_message(**overrides): + message = { + "message_id": 100, + "date": 1700000000, + "from": dict(ALICE), + "chat": {"id": 30, "type": "private", "first_name": "Alice"}, + "text": "hello", + } + return {**message, **overrides} + + +def telegram_update(update_id=1, **overrides): + return {"update_id": update_id, "message": telegram_message(**overrides)} + + +def mention(text, entity_text, kind="mention", **extra): + before = text[: text.index(entity_text)] + return { + "type": kind, + "offset": len(before.encode("utf-16-le")) // 2, + "length": len(entity_text.encode("utf-16-le")) // 2, + **extra, + } + + +def group_update(text="@agentchat_bot hi", entities=None, update_id=1, **overrides): + if entities is None: + entities = [mention(text, "@agentchat_bot")] if "@agentchat_bot" in text else [] + return telegram_update(update_id, chat=dict(GROUP), text=text, entities=entities, **overrides) + + +def callback_update(data="yes", update_id=2, **overrides): + keyboard = {"inline_keyboard": [[ + {"text": "Yes please", "callback_data": "yes"}, + {"text": "No", "callback_data": "no"}, + ]]} + callback = { + "id": "cb-1", + "from": dict(ALICE), + "chat_instance": "instance", + "data": data, + "message": { + "message_id": 55, + "date": 1700000000, + "from": dict(BOT), + "chat": {"id": 30, "type": "private", "first_name": "Alice"}, + "text": "Continue?", + "reply_markup": keyboard, + }, + } + return {"update_id": update_id, "callback_query": {**callback, **overrides}} + + +def read(update): + return normalize(update, bot_id=99, bot_username="agentchat_bot") + + +def body(update_id, **overrides): + return json.dumps(telegram_update(update_id, **overrides)) + + +async def until(predicate): + for _ in range(600): + if predicate(): + return + await asyncio.sleep(0.005) + raise AssertionError("Condition was not reached") + + +async def nothing(channel, message): + pass + + +def fast_retries(monkeypatch): + monkeypatch.setattr(telegram_adapter, "RETRY_BASE", 0.01) + monkeypatch.setattr(telegram_adapter, "BUSY_DELAY", 0.02) + + +def served(api, **options): + client = TelegramBotClient(TOKEN, api_root=api.root) + return Telegram(bot_token=TOKEN, client=client, **{"shutdown_timeout": 1, **options}) + + +def faked(client=None, **options): + return Telegram(bot_token="99:fake", client=client or TelegramClient(), **options) + + +def hooked(client=None, **options): + return faked(client, webhook_url=HOOK_URL, webhook_secret=SECRET, **options) + + +class TelegramClient: + def __init__(self, receipt=None, *, batches=None): + self.receipt = receipt if receipt is not None else Receipt(101) + self.batches = list(batches or []) + self.calls = [] + self.uploads = [] + self.started, self.stopped = asyncio.Event(), asyncio.Event() + self.closed = False + + def method_calls(self, method): + return [payload for name, payload in self.calls if name == method] + + async def call(self, method, payload=None, **options): + self.calls.append((method, payload)) + if method == "getMe": + return dict(BOT) + if method == "getUpdates": + if payload.get("timeout") == 0: + return [] + if self.batches: + return self.batches.pop(0) + self.started.set() + await self.stopped.wait() + return [] + if method == "sendMessage": + return {"message_id": self.receipt.get(method, payload)} + return True + + async def upload(self, *, method, fields, file_field, filename, content): + self.uploads.append((method, fields, file_field, filename, content)) + return {"message_id": 200 + len(self.uploads), + "document": {"file_id": f"file-{len(self.uploads)}"}} + + async def download_file(self, file_id, *, max_bytes): + self.calls.append(("download_file", {"file_id": file_id, "max_bytes": max_bytes})) + return f"documents/{file_id}", b"content" + + async def close(self): + self.closed = True + self.stopped.set() + + +def _ok(result): + return web.json_response({"ok": True, "result": result}) + + +def _error(code, description, retry_after=None): + body = {"ok": False, "error_code": code, "description": description} + if retry_after: + body["parameters"] = {"retry_after": retry_after} + return web.json_response(body, status=code) + + +class FakeBotAPI: + def __init__(self, token): + self.token = token + self.bot = {**BOT, "id": int(token.split(":")[0])} + self.updates = [] + self.next_id = 100 + self.calls = [] + self.offsets = [] + self.webhook = None + self.files = {} + self.failures = {} + self.poll_started = asyncio.Event() + self._changed = asyncio.Condition() + self.server = None + self.root = None + + async def __aenter__(self): + app = web.Application() + app.router.add_post("/bot{token}/{method}", self._handle) + app.router.add_get("/file/bot{token}/{path:.+}", self._download) + self.server = TestServer(app) + await self.server.start_server() + self.root = str(self.server.make_url("")).rstrip("/") + return self + + async def __aexit__(self, *args): + await self.server.close() + + async def push(self, update): + update = {**update, "update_id": self.next_id} + self.next_id += 1 + self.updates.append(update) + async with self._changed: + self._changed.notify_all() + return update["update_id"] + + def method_calls(self, method): + return [payload for name, payload in self.calls if name == method] + + async def _payload(self, request): + if request.content_type == "multipart/form-data": + fields = {} + async for part in await request.multipart(): + if part.filename: + fields[part.name] = (part.filename, await part.read()) + else: + fields[part.name] = await part.text() + return fields + return await request.json() if request.can_read_body else {} + + async def _handle(self, request): + method = request.match_info["method"] + payload = await self._payload(request) + self.calls.append((method, payload)) + if request.match_info["token"] != self.token: + return _error(401, "Unauthorized") + scripted = self.failures.get(method) + if scripted: + status, body = scripted.pop(0) + if body is None: + return web.Response(status=status, text="bad gateway") + return web.json_response(body, status=status) + if method == "getMe": + return _ok(self.bot) + if method == "deleteWebhook": + self.webhook = None + return _ok(True) + if method == "setWebhook": + self.webhook = payload + return _ok(True) + if method == "getUpdates": + return await self._get_updates(payload) + if method == "sendMessage": + return _ok({"message_id": 500 + len(self.method_calls("sendMessage")), + "chat": {"id": payload["chat_id"]}, "text": payload["text"]}) + if method == "sendDocument": + return _ok({"message_id": 600, "document": {"file_id": "uploaded-file"}}) + if method == "getFile": + file = self.files.get(payload["file_id"]) + return _ok(file["info"]) if file else _error(400, "Bad Request: invalid file_id") + if method in ("sendChatAction", "answerCallbackQuery"): + return _ok(True) + return _error(404, "Not Found") + + async def _get_updates(self, payload): + if self.webhook is not None: + return _error(409, "Conflict: can't use getUpdates method while webhook is active") + offset = payload.get("offset") + self.offsets.append(offset) + if offset is not None: + self.updates = [u for u in self.updates if u["update_id"] >= offset] + limit = payload.get("limit", 100) + self.poll_started.set() + async with self._changed: + try: + await asyncio.wait_for( + self._changed.wait_for(lambda: bool(self.updates)), + timeout=payload.get("timeout", 0) or 0.001, + ) + except TimeoutError: + pass + return _ok(self.updates[:limit]) + + async def _download(self, request): + if request.match_info["token"] != self.token: + return web.Response(status=404) + for file in self.files.values(): + if file["info"].get("file_path") == request.match_info["path"] and file["content"]: + return web.Response(body=file["content"]) + return web.Response(status=404) From 98de15761e6834650cea3eb64392ed22657cbc6f Mon Sep 17 00:00:00 2001 From: CatalanCabbage Date: Wed, 14 Oct 2026 02:06:32 +0530 Subject: [PATCH 3/5] docs: Telegram setup guide and echo example --- README.md | 3 +- docs/architecture.md | 6 +- docs/providers.md | 10 ++- docs/telegram.md | 142 ++++++++++++++++++++++++++++++++++++++ examples/telegram_echo.py | 26 +++++++ 5 files changed, 182 insertions(+), 5 deletions(-) create mode 100644 docs/telegram.md create mode 100644 examples/telegram_echo.py diff --git a/README.md b/README.md index 99cd159..643826f 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # agentchat -a minimal sdk to connect your agents to slack, teams, imessage, whatsapp. +a minimal sdk to connect your agents to slack, teams, imessage, whatsapp, telegram. use any SDK (Claude Agents, OpenAI Agents, Langchain DeepAgents, Pydantic Agents) and instantly connect it to all supported channels. @@ -187,3 +187,4 @@ async def respond(context: MessageContext) -> str: | iMessage | BlueBubbles on a Mac signed into Messages | `imessage` | [iMessage guide](docs/imessage.md) | | WhatsApp | Business Cloud API with signed webhooks | `whatsapp` | [WhatsApp guide](docs/whatsapp.md) | | Discord | Bot gateway | `discord` | [Discord guide](docs/discord.md) | +| Telegram | Bot API long polling or HTTPS webhook | `telegram` | [Telegram guide](docs/telegram.md) | diff --git a/docs/architecture.md b/docs/architecture.md index 161cf15..aa26e4c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -77,9 +77,9 @@ channels/ Every Slack-specific module lives below `slack/`. There are no sibling Slack compatibility modules or provider aliases in the shared namespace. -Discord, Teams, iMessage, and WhatsApp use the same package boundary with +Discord, Teams, iMessage, WhatsApp, and Telegram use the same package boundary with `adapter.py`, normalization in `events.py`, and text validation in `rendering.py`. -Discord and BlueBubbles have small REST transport modules too. The provider-neutral +Discord, BlueBubbles, and Telegram have small REST transport modules too. The provider-neutral private helper `agentchat/_delivery.py` shares bounded background dispatch and cleanup. Provider authentication, acknowledgement, and overflow policy stay in each adapter. See the [provider guides](providers.md) for supported operations. @@ -149,5 +149,5 @@ not import Slack, aiohttp or OpenAI Agents. Installation defaults are unchanged: Slack dependencies remain installed with the base package, while OpenAI Agents still uses the `openai` extra. -Discord, Teams, and iMessage use separate optional extras. WhatsApp uses aiohttp +Discord, Teams, and iMessage use separate optional extras. WhatsApp and Telegram use aiohttp from the base install. Requesting one provider does not import the others. diff --git a/docs/providers.md b/docs/providers.md index 8d36309..da1c222 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -10,9 +10,10 @@ a runnable echo example; no agent framework or model key is needed for that chec | Teams | `pip install -e '.[teams]'` | Authenticated HTTPS endpoint | [Teams](teams.md) | | iMessage | `pip install -e '.[imessage]'` | BlueBubbles Socket.IO + REST | [iMessage](imessage.md) | | WhatsApp | `pip install -e '.[whatsapp]'` | Signed HTTPS webhook + Cloud API | [WhatsApp](whatsapp.md) | +| Telegram | `pip install -e '.[telegram]'` | Bot API long polling or HTTPS webhook | [Telegram](telegram.md) | Use each provider's own package for imports, following the current Slack guide. -The `whatsapp` extra needs +The `whatsapp` and `telegram` extras need no additional dependency because aiohttp is already installed. Provider packages and agent-framework integrations load only when imported. Python 3.11+ is required. @@ -83,6 +84,10 @@ These four new adapters implement the minimal `Channel` contract: receive, lifecycle, and confirmed text replies. They normalize sender identity and incoming attachment references. They do not download those attachments. +Telegram also implements `set_status()` (the typing indicator), `reply_rich()` +(native reply markup such as inline keyboards), `upload_files()`, and +`download_attachment()`. It does not implement native history or mirroring. + History uses the application's local `State`; default `MemoryState` is lost on restart. Native history, working indicators, rich replies, file transfers, and cross-surface mirroring remain optional capabilities. On the new adapters, @@ -104,6 +109,9 @@ conversation using the configured `State`. - Teams returns 202 after admission; WhatsApp returns 200. A full HTTP receiver returns 503 before admission so the provider can apply its retry policy. WhatsApp batches are admitted atomically. +- Telegram webhooks return 503 the same way. Telegram polling stops reading at + the first update it cannot admit and polls again from it, so Telegram redelivers + it instead of the adapter dropping it. - Discord and BlueBubbles callbacks cannot ask the gateway to replay an event. At capacity, the adapter logs a warning and drops the excess event. Reconnects do not provide durable catch-up. Size the limit/host for the expected workload. diff --git a/docs/telegram.md b/docs/telegram.md new file mode 100644 index 0000000..f5e2b63 --- /dev/null +++ b/docs/telegram.md @@ -0,0 +1,142 @@ +# Telegram + +Connect a Telegram bot through the Bot API. Long polling is the default, so the +first check needs no public URL, tunnel, or model API key. Webhook mode is +available for deployments that already serve HTTPS. + +## Get the first reply + +1. In Telegram, open [@BotFather](https://t.me/BotFather), send `/newbot`, and + choose a display name and a username ending in `bot`. BotFather replies with a + token of the form `123456789:AA...`. Store it as `TELEGRAM_BOT_TOKEN`; the part + before the colon is the bot's numeric ID. +2. From this repository, install and run: + + ```bash + python -m pip install -e '.[telegram]' + export TELEGRAM_BOT_TOKEN='123456789:your-bot-token' + python examples/telegram_echo.py + ``` + + On start the adapter checks the token with `getMe`, removes any webhook + (`deleteWebhook` without dropping pending updates), and long-polls `getUpdates`. +3. Open a private chat with your bot, press **Start**, and send `hello`. Expect + `hello` back as a reply to your message. Stop with Ctrl+C. + +The example echoes each accepted input. Replace its handler with your runner +after authorizing the sender. See [shared usage](providers.md) to run several +adapters or bind an existing runner directly. + +### Groups and privacy mode + +Add the bot to a group, then address it with `@yourbot question`, a command such +as `/status@yourbot`, or a reply to one of its messages. Bots have privacy mode +enabled by default, so Telegram only delivers those addressed messages to them +anyway. The adapter applies the same rule itself: unaddressed group messages never +reach the agent, even if you disable privacy mode with BotFather's `/setprivacy` +or make the bot a group admin. Privacy mode changes apply to groups the bot joins +afterwards; remove and re-add the bot to an existing group. + +## Webhook mode + +Set both variables to switch from polling to a webhook: + +```bash +export TELEGRAM_WEBHOOK_URL='https://YOUR_HOST/webhooks/telegram' +export TELEGRAM_WEBHOOK_SECRET="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')" +python examples/telegram_echo.py +``` + +The URL must use HTTPS and end in `/webhooks/telegram`. The secret may use +`A-Z`, `a-z`, `0-9`, `_` and `-`, up to 256 characters. The adapter listens on +`127.0.0.1:3980`, then calls `setWebhook` with the URL and secret. Forward a +public HTTPS tunnel/proxy to this port. Containers can set `TELEGRAM_HOST=0.0.0.0`; +`TELEGRAM_PORT` changes the port. Telegram sends each update with an +`X-Telegram-Bot-Api-Secret-Token` header; requests without the matching secret +get 401. + +`close()` leaves the webhook set, so Telegram keeps queueing updates while the bot +is down and redelivers them after a restart. Telegram allows one consumer per +token: starting a poller with the same token removes the webhook. + +For an existing aiohttp server, construct `Telegram(bot_token=..., webhook_url=..., +webhook_secret=...)`, host `telegram.create_app()` yourself, and bind the receiver +before serving requests. Register the webhook yourself, for example: + +```bash +curl -s "https://api.telegram.org/bot$TELEGRAM_BOT_TOKEN/setWebhook" \ + -d "url=$TELEGRAM_WEBHOOK_URL" -d "secret_token=$TELEGRAM_WEBHOOK_SECRET" \ + -d 'allowed_updates=["message","callback_query"]' +``` + +The handler looks up the bot with `getMe` on the first update and returns 503 +until that succeeds. Call `await telegram.close()` at shutdown; do not also call +`telegram.run()`. + +## Routing and limits + +- Each private chat and group is a conversation. In a forum supergroup, each topic + is its own conversation and sets `message.thread_id` to the topic ID. Message + IDs have the form `telegram:::`; button taps use + `telegram::callback:`. +- Messages from bots, channel posts, edits, service messages, and messages without + text, a caption, or a file are ignored. The bot's `@username` is removed from the + text, and from a command suffix (`/status@yourbot` becomes `/status`). +- Photos (largest size), documents, audio, video, voice notes, video notes, + animations, and stickers become attachment references; captions become the + message text. `download_attachment()` fetches the bytes on demand. The Bot API + serves files up to 20 MB. +- Replies quote the original message, post in the same topic, and disable link + previews. If the original was deleted, the reply still posts. +- Agent Markdown is converted to Telegram message entities, never `parse_mode`: + bold (`**`), italic (`*` or `_`), strikethrough (`~~`), spoiler (`||`), inline + code, fenced code blocks with a language, `http(s)` and `tg` links, headings + (bold), bullet and numbered lists, and blockquotes. `__` and intra-word `_`/`*` + stay literal, so `snake_case` and `__init__.py` are safe. +- A reply must fit 4,096 UTF-16 code units after conversion. Oversized replies + raise `ValueError` before sending. Split longer output into separate + `context.reply()` calls. + +Replies, uploads, and status updates make one attempt; errors propagate. A Bot API +error raises `TelegramError` (a `RuntimeError`) with `error_code`, `description`, +and, on HTTP 429, `retry_after` seconds for your host's rate-limit scheduling. +Polling itself retries network errors, HTTP 5xx, and 429 with backoff; other +errors, such as 401 or 409, stop `run()`. + +Telegram implements the optional capabilities: + +- `set_status()` shows the typing indicator. Telegram clears it after about five + seconds and has no custom text, so the adapter re-sends it every four seconds + until you clear it with `""`, two minutes pass without another call, or the + adapter closes. `context.working()` refreshes it for the duration of a handler. +- `reply_rich()` sends `RichReply.text` with one native Telegram reply markup in + `blocks`, such as an inline keyboard: + `RichReply(text="Continue?", blocks=({"inline_keyboard": [[{"text": "Yes", "callback_data": "yes"}]]},))`. + Attachments are not supported. +- Inline keyboard taps arrive as ordinary messages whose text is the tapped + button's label; `metadata["callback_data"]` holds its data. The adapter answers + the tap so the button's spinner stops. +- `upload_files()` sends each file as a document to the source chat or topic, with + `title` as the caption. The Bot API accepts uploads up to 50 MB. If a later file + fails, earlier files were already delivered. + +## Troubleshooting + +Store the token as a secret. Revoke and regenerate it with BotFather's `/revoke` +if it leaks, then restart. + +| Symptom | Check | +| --- | --- | +| `run()` fails with authentication failed | Copy the full token from BotFather, including the numeric prefix. | +| `run()` fails with HTTP 409 Conflict | Another process polls with the same token, or something set a webhook for it. Stop the other consumer. | +| The bot ignores group messages | Mention it, use `/command@yourbot`, or reply to its message. | +| A webhook never receives updates | Check the HTTPS URL and tunnel, and run `getWebhookInfo` for Telegram's last delivery error. | +| Webhook POST returns 401 | `TELEGRAM_WEBHOOK_SECRET` must match the `secret_token` registered with `setWebhook`. | +| HTTP 503 from this receiver | The bot identity is not yet known or the pending-work limit is full; Telegram retries. | +| Replies fail with HTTP 403 | The user blocked the bot, or the bot was removed from the group. | +| HTTP 429 | Pause sends for `TelegramError.retry_after` seconds in your host. | + +Local tests exercise polling, webhooks, and sends against a simulated Bot API. +A live echo with your own bot remains the final deployment check. + +Reference: [Telegram Bot API](https://core.telegram.org/bots/api). diff --git a/examples/telegram_echo.py b/examples/telegram_echo.py new file mode 100644 index 0000000..03546e5 --- /dev/null +++ b/examples/telegram_echo.py @@ -0,0 +1,26 @@ +"""Echo messages after completing docs/telegram.md; no model API key required.""" + +import asyncio +import logging + +from agentchat import AgentChat, MessageContext +from agentchat.channels.telegram import Telegram + + +async def main() -> None: + app = AgentChat(channels=[Telegram.from_env()]) + + @app.on_message + async def respond(context: MessageContext) -> str: + # Authorize the sender here before replacing this echo with an agent. + return context.message.text or "Received your attachment." + + try: + await app.run() + finally: + await app.close() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + asyncio.run(main()) From c85a11f27ce3baa753e2354fb165eb8858434078 Mon Sep 17 00:00:00 2001 From: CatalanCabbage Date: Wed, 14 Oct 2026 02:12:24 +0530 Subject: [PATCH 4/5] Telegram: name constants and results, drop restating comments --- examples/telegram_echo.py | 2 - pyproject.toml | 2 +- src/agentchat/channels/telegram/__init__.py | 2 - src/agentchat/channels/telegram/adapter.py | 146 +++++++++++-------- src/agentchat/channels/telegram/events.py | 141 ++++++++++-------- src/agentchat/channels/telegram/rendering.py | 83 ++++++----- src/agentchat/channels/telegram/transport.py | 29 +++- tests/channels/telegram/test_polling.py | 5 +- tests/support/telegram.py | 4 +- 9 files changed, 241 insertions(+), 173 deletions(-) diff --git a/examples/telegram_echo.py b/examples/telegram_echo.py index 03546e5..a2525c2 100644 --- a/examples/telegram_echo.py +++ b/examples/telegram_echo.py @@ -1,5 +1,3 @@ -"""Echo messages after completing docs/telegram.md; no model API key required.""" - import asyncio import logging diff --git a/pyproject.toml b/pyproject.toml index aa84c5d..9ecc557 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ teams = [ ] imessage = ["python-socketio[asyncio-client]>=5.13,<6"] whatsapp = [] # Uses the existing aiohttp dependency. -telegram = [] # Uses the existing aiohttp dependency. +telegram = [] dev = [ "openai-agents>=0.4,<1", "discord.py>=2.6,<3", diff --git a/src/agentchat/channels/telegram/__init__.py b/src/agentchat/channels/telegram/__init__.py index 3ef36c4..3a04788 100644 --- a/src/agentchat/channels/telegram/__init__.py +++ b/src/agentchat/channels/telegram/__init__.py @@ -1,5 +1,3 @@ -"""Telegram adapter and provider-local helpers, loaded on demand.""" - from typing import TYPE_CHECKING if TYPE_CHECKING: diff --git a/src/agentchat/channels/telegram/adapter.py b/src/agentchat/channels/telegram/adapter.py index 4d88808..ecd9625 100644 --- a/src/agentchat/channels/telegram/adapter.py +++ b/src/agentchat/channels/telegram/adapter.py @@ -17,12 +17,32 @@ from agentchat._delivery import DeliveryTasks from agentchat.channels.base import MessageReceiver -from agentchat.channels.telegram.events import normalize +from agentchat.channels.telegram.events import ( + METADATA_BOT_ID, + METADATA_CALLBACK_DATA, + METADATA_CALLBACK_QUERY_ID, + METADATA_CHAT_ID, + METADATA_MESSAGE_ID, + METADATA_THREAD_ID, + TELEGRAM_CHANNEL, + UPDATE_CALLBACK_QUERY, + UPDATE_MESSAGE, + normalize, +) from agentchat.channels.telegram.rendering import text_payload, truncate_caption from agentchat.channels.telegram.transport import ( + METHOD_ANSWER_CALLBACK_QUERY, + METHOD_DELETE_WEBHOOK, + METHOD_GET_ME, + METHOD_GET_UPDATES, + METHOD_SEND_CHAT_ACTION, + METHOD_SEND_DOCUMENT, + METHOD_SEND_MESSAGE, + METHOD_SET_WEBHOOK, REQUEST_TIMEOUT, TelegramBotClient, TelegramError, + TelegramFile, ) from agentchat.models import ( Attachment, @@ -36,23 +56,30 @@ logger = logging.getLogger(__name__) +ENV_BOT_TOKEN = "TELEGRAM_BOT_TOKEN" +ENV_WEBHOOK_URL = "TELEGRAM_WEBHOOK_URL" +ENV_WEBHOOK_SECRET = "TELEGRAM_WEBHOOK_SECRET" +ENV_HOST = "TELEGRAM_HOST" +ENV_PORT = "TELEGRAM_PORT" +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 3980 POLL_TIMEOUT = 30 -# The HTTP timeout must outlive Telegram's long-poll hold or every empty poll times out. -POLL_HTTP_MARGIN = 10 -ALLOWED_UPDATES = ["message", "callback_query"] +POLL_REQUEST_TIMEOUT = POLL_TIMEOUT + 10 +ALLOWED_UPDATES = [UPDATE_MESSAGE, UPDATE_CALLBACK_QUERY] WEBHOOK_PATH = "/webhooks/telegram" SECRET_HEADER = "X-Telegram-Bot-Api-Secret-Token" RETRY_BASE = 1.0 RETRY_MAX = 30.0 BUSY_DELAY = 1.0 -CONFIRM_TIMEOUT = 2 -ANSWER_TIMEOUT = 5 +OFFSET_CONFIRM_TIMEOUT = 2 +CALLBACK_ANSWER_TIMEOUT = 5 # Telegram clears a chat action after about five seconds. TYPING_INTERVAL = 4 TYPING_EXPIRY = 120 +CHAT_ACTION_TYPING = "typing" BOT_TOKEN_PATTERN = re.compile(r"([0-9]+):[A-Za-z0-9_-]+") WEBHOOK_SECRET_PATTERN = re.compile(r"[A-Za-z0-9_-]{1,256}") -CALLBACK_ONLY_METADATA = frozenset({"callback_query_id", "callback_data"}) +CALLBACK_ONLY_METADATA = frozenset({METADATA_CALLBACK_QUERY_ID, METADATA_CALLBACK_DATA}) class BotAPIClient(Protocol): @@ -68,7 +95,7 @@ async def upload( self, *, method: str, fields: dict, file_field: str, filename: str, content: bytes, ) -> Any: ... - async def download_file(self, file_id: str, *, max_bytes: int) -> tuple[str, bytes]: ... + async def download_file(self, file_id: str, *, max_bytes: int) -> TelegramFile: ... async def close(self) -> None: ... @@ -79,7 +106,7 @@ class _Destination(NamedTuple): thread_id: int | None -def retry_delay(attempt: int, retry_after: int | None) -> float: +def retry_delay(*, attempt: int, retry_after: int | None) -> float: if retry_after: return float(retry_after) return min(RETRY_BASE * 2**attempt, RETRY_MAX) @@ -97,9 +124,7 @@ def _is_plain_filename(filename: object) -> bool: class Telegram: - """Telegram Bot API: long polling by default, or a secret-checked webhook.""" - - name = "telegram" + name = TELEGRAM_CHANNEL def __init__( self, @@ -107,8 +132,8 @@ def __init__( bot_token: str, webhook_url: str | None = None, webhook_secret: str | None = None, - host: str = "127.0.0.1", - port: int = 3980, + host: str = DEFAULT_HOST, + port: int = DEFAULT_PORT, client: BotAPIClient | None = None, max_pending_events: int = 64, shutdown_timeout: float = 20, @@ -144,21 +169,21 @@ def __init__( @classmethod def from_env(cls) -> Telegram: - token = os.getenv("TELEGRAM_BOT_TOKEN") + token = os.getenv(ENV_BOT_TOKEN) if not token: - raise RuntimeError("TELEGRAM_BOT_TOKEN is required") - webhook_url = os.getenv("TELEGRAM_WEBHOOK_URL") or None - webhook_secret = os.getenv("TELEGRAM_WEBHOOK_SECRET") or None + raise RuntimeError(f"{ENV_BOT_TOKEN} is required") + webhook_url = os.getenv(ENV_WEBHOOK_URL) or None + webhook_secret = os.getenv(ENV_WEBHOOK_SECRET) or None if (webhook_url is None) != (webhook_secret is None): raise RuntimeError( - "TELEGRAM_WEBHOOK_URL and TELEGRAM_WEBHOOK_SECRET are required together" + f"{ENV_WEBHOOK_URL} and {ENV_WEBHOOK_SECRET} are required together" ) return cls( bot_token=token, webhook_url=webhook_url, webhook_secret=webhook_secret, - host=os.getenv("TELEGRAM_HOST", "127.0.0.1"), - port=int(os.getenv("TELEGRAM_PORT", "3980")), + host=os.getenv(ENV_HOST, DEFAULT_HOST), + port=int(os.getenv(ENV_PORT, str(DEFAULT_PORT))), ) def bind(self, receiver: MessageReceiver) -> None: @@ -209,9 +234,9 @@ async def close(self) -> None: finally: await self._client.close() - async def _identify(self) -> None: + async def _load_bot_identity(self) -> None: try: - me = await self._client.call("getMe") + me = await self._client.call(METHOD_GET_ME) except TelegramError as error: if error.is_transient: raise @@ -228,9 +253,9 @@ async def _identify(self) -> None: self._username = username async def _serve(self) -> None: - await self._identify() + await self._load_bot_identity() if self._webhook_url is None: - await self._client.call("deleteWebhook", {"drop_pending_updates": False}) + await self._client.call(METHOD_DELETE_WEBHOOK, {"drop_pending_updates": False}) await self._poll() else: await self._serve_webhook() @@ -245,28 +270,28 @@ async def _poll(self) -> None: payload["offset"] = self._offset try: updates = await self._client.call( - "getUpdates", payload, request_timeout=POLL_TIMEOUT + POLL_HTTP_MARGIN + METHOD_GET_UPDATES, payload, request_timeout=POLL_REQUEST_TIMEOUT ) except TelegramError as error: if not error.is_transient: raise - delay = retry_delay(attempt, error.retry_after) + delay = retry_delay(attempt=attempt, retry_after=error.retry_after) logger.warning("Telegram polling failed (%s); retrying in %.1fs", error, delay) attempt += 1 await asyncio.sleep(delay) continue attempt = 0 self._confirmed_offset = self._offset - if not self._admit(updates if isinstance(updates, list) else []): + if not self._has_admitted_all(updates if isinstance(updates, list) else []): await asyncio.sleep(BUSY_DELAY) - def _admit(self, updates: list) -> bool: + def _has_admitted_all(self, updates: list) -> bool: for update in updates: update_id = update.get("update_id") if isinstance(update, dict) else None if not isinstance(update_id, int): continue message = self._normalize(update) - if message is not None and not self._submit(message): + if message is not None and not self._has_submitted(message): return False self._offset = update_id + 1 return True @@ -275,10 +300,10 @@ def _normalize(self, update: dict) -> Message | None: assert self._username is not None return normalize(update, bot_id=self._bot_id, bot_username=self._username) - def _submit(self, message: Message) -> bool: + def _has_submitted(self, message: Message) -> bool: if not self._delivery.submit(self, message): return False - callback_id = message.metadata.get("callback_query_id") + callback_id = message.metadata.get(METADATA_CALLBACK_QUERY_ID) if isinstance(callback_id, str): task = asyncio.create_task(self._answer_callback(callback_id)) self._answers.add(task) @@ -288,9 +313,9 @@ def _submit(self, message: Message) -> bool: async def _answer_callback(self, callback_id: str) -> None: try: await self._client.call( - "answerCallbackQuery", + METHOD_ANSWER_CALLBACK_QUERY, {"callback_query_id": callback_id}, - request_timeout=ANSWER_TIMEOUT, + request_timeout=CALLBACK_ANSWER_TIMEOUT, ) except Exception as error: logger.warning("Telegram callback answer failed (%s)", error) @@ -301,9 +326,9 @@ async def _confirm_offset(self) -> None: return try: await self._client.call( - "getUpdates", + METHOD_GET_UPDATES, {"offset": self._offset, "limit": 1, "timeout": 0}, - request_timeout=CONFIRM_TIMEOUT, + request_timeout=OFFSET_CONFIRM_TIMEOUT, ) except TelegramError as error: logger.warning("Telegram offset confirmation failed (%s)", error) @@ -314,7 +339,7 @@ async def _serve_webhook(self) -> None: await web.TCPSite(self._runner, self._host, self._port).start() # Listen first: Telegram starts sending as soon as setWebhook succeeds. await self._client.call( - "setWebhook", + METHOD_SET_WEBHOOK, { "url": self._webhook_url, "secret_token": self._webhook_secret, @@ -325,8 +350,8 @@ async def _serve_webhook(self) -> None: async def _receive(self, request: web.Request) -> web.Response: assert self._webhook_secret is not None - supplied = request.headers.get(SECRET_HEADER, "") - if not hmac.compare_digest(supplied.encode(), self._webhook_secret.encode()): + supplied_secret = request.headers.get(SECRET_HEADER, "") + if not hmac.compare_digest(supplied_secret.encode(), self._webhook_secret.encode()): raise web.HTTPUnauthorized(text="Invalid webhook secret") try: update = json.loads(await request.read()) @@ -338,30 +363,33 @@ async def _receive(self, request: web.Request) -> web.Response: raise web.HTTPServiceUnavailable(text="Telegram receiver is closed") if self._username is None: try: - await self._identify() + await self._load_bot_identity() except Exception as error: logger.warning("Telegram bot identity unavailable (%s)", error) raise web.HTTPServiceUnavailable(text="Telegram receiver is starting") from None message = self._normalize(update) - if message is not None and not self._submit(message): + if message is not None and not self._has_submitted(message): raise web.HTTPServiceUnavailable(text="Telegram receiver is busy") return web.Response(text="OK") def _destination(self, source: Message) -> _Destination: - data = source.metadata - chat_id, message_id = data.get("chat_id"), data.get("message_id") - thread = data.get("message_thread_id") + metadata = source.metadata + chat_id, message_id = metadata.get(METADATA_CHAT_ID), metadata.get(METADATA_MESSAGE_ID) + thread_id = metadata.get(METADATA_THREAD_ID) if ( source.channel != self.name - or data.get("bot_id") != self._bot_id + or metadata.get(METADATA_BOT_ID) != self._bot_id or not isinstance(chat_id, int) or not isinstance(message_id, int) or isinstance(chat_id, bool) or isinstance(message_id, bool) - or (thread is not None and (not isinstance(thread, int) or isinstance(thread, bool))) + or ( + thread_id is not None + and (not isinstance(thread_id, int) or isinstance(thread_id, bool)) + ) ): raise ValueError("Source does not belong to this Telegram adapter") - return _Destination(chat_id, message_id, thread) + return _Destination(chat_id=chat_id, message_id=message_id, thread_id=thread_id) async def _send( self, *, source: Message, destination: _Destination, payload: dict, text: str, @@ -376,7 +404,7 @@ async def _send( } if destination.thread_id is not None: payload["message_thread_id"] = destination.thread_id - result = await self._client.call("sendMessage", payload) + result = await self._client.call(METHOD_SEND_MESSAGE, payload) sent = result.get("message_id") if isinstance(result, dict) else None if not isinstance(sent, int) or isinstance(sent, bool) or sent <= 0: raise RuntimeError("Telegram did not return a message receipt") @@ -385,14 +413,14 @@ async def _send( if key not in CALLBACK_ONLY_METADATA } return Message( - id=f"telegram:{self._bot_id}:{destination.chat_id}:{sent}", + id=f"{TELEGRAM_CHANNEL}:{self._bot_id}:{destination.chat_id}:{sent}", conversation_id=source.conversation_id, channel=self.name, sender=Sender(id=str(self._bot_id), display_name=self._first_name), text=text, role="assistant", thread_id=source.thread_id, - metadata=MappingProxyType({**metadata, "message_id": sent}), + metadata=MappingProxyType({**metadata, METADATA_MESSAGE_ID: sent}), ) async def reply(self, source: Message, content: str) -> Message: @@ -420,10 +448,12 @@ async def set_status(self, source: Message, status: str) -> bool: previous.cancel() if not status: return True - payload: dict[str, object] = {"chat_id": destination.chat_id, "action": "typing"} + payload: dict[str, object] = { + "chat_id": destination.chat_id, "action": CHAT_ACTION_TYPING, + } if destination.thread_id is not None: payload["message_thread_id"] = destination.thread_id - await self._client.call("sendChatAction", payload) + await self._client.call(METHOD_SEND_CHAT_ACTION, payload) # A concurrent call for the same chat may have stored a task while this one awaited. if previous := self._typing.pop(key, None): previous.cancel() @@ -438,7 +468,7 @@ async def _keep_typing(self, key: tuple[int, int | None], payload: dict) -> None await asyncio.sleep(TYPING_INTERVAL) with contextlib.suppress(Exception): await self._client.call( - "sendChatAction", payload, request_timeout=TYPING_INTERVAL, + METHOD_SEND_CHAT_ACTION, payload, request_timeout=TYPING_INTERVAL, ) finally: if self._typing.get(key) is asyncio.current_task(): @@ -467,7 +497,7 @@ async def upload_files( if file.title: fields["caption"] = truncate_caption(file.title) result = await self._client.upload( - method="sendDocument", + method=METHOD_SEND_DOCUMENT, fields=fields, file_field="document", filename=file.filename, @@ -490,12 +520,12 @@ async def download_attachment( raise ValueError("Attachment does not belong to the source message") if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or max_bytes <= 0: raise ValueError("Expected a positive attachment byte limit") - path, content = await self._client.download_file(attachment.id, max_bytes=max_bytes) - if not content: + file = await self._client.download_file(attachment.id, max_bytes=max_bytes) + if not file.content: raise ValueError("This Telegram file was empty. Send the attachment again.") return DownloadedFile( id=attachment.id, - name=attachment.name or posixpath.basename(path) or "attachment", - content=content, + name=attachment.name or posixpath.basename(file.path) or "attachment", + content=file.content, media_type=attachment.media_type, ) diff --git a/src/agentchat/channels/telegram/events.py b/src/agentchat/channels/telegram/events.py index ab687b4..3cc6685 100644 --- a/src/agentchat/channels/telegram/events.py +++ b/src/agentchat/channels/telegram/events.py @@ -1,11 +1,20 @@ -"""Normalize Telegram Bot API updates; no network calls.""" - from __future__ import annotations from types import MappingProxyType +from typing import NamedTuple from agentchat.models import Attachment, Message, Sender +TELEGRAM_CHANNEL = "telegram" +UPDATE_MESSAGE = "message" +UPDATE_CALLBACK_QUERY = "callback_query" +METADATA_BOT_ID = "bot_id" +METADATA_CHAT_ID = "chat_id" +METADATA_CHAT_TYPE = "chat_type" +METADATA_MESSAGE_ID = "message_id" +METADATA_THREAD_ID = "message_thread_id" +METADATA_CALLBACK_QUERY_ID = "callback_query_id" +METADATA_CALLBACK_DATA = "callback_data" CHAT_PRIVATE = "private" GROUP_CHATS = frozenset({"group", "supergroup"}) ENTITY_MENTION = "mention" @@ -16,60 +25,71 @@ FILE_KINDS = ("animation", "document", "audio", "video", "voice", "video_note", "sticker") -def _mapping(value: object) -> dict: +class _Span(NamedTuple): + start: int + end: int + text: str + + +class _BotAddress(NamedTuple): + is_addressed: bool + text: str + + +def _as_dict(value: object) -> dict: return value if isinstance(value, dict) else {} -def _items(value: object) -> list: +def _as_list(value: object) -> list: return value if isinstance(value, list) else [] -def _is_id(value: object) -> bool: +def _is_non_bool_int(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool) def _sender(user: dict) -> Sender | None: - if not _is_id(user.get("id")) or user.get("is_bot") is not False: + if not _is_non_bool_int(user.get("id")) or user.get("is_bot") is not False: return None names = (user.get("first_name"), user.get("last_name")) display_name = " ".join(name for name in names if isinstance(name, str) and name) return Sender(id=str(user["id"]), display_name=display_name or None) -def _span(units: bytes, entity: dict) -> tuple[int, int, str] | None: +def _entity_span(units: bytes, entity: dict) -> _Span | None: offset, length = entity.get("offset"), entity.get("length") - if not _is_id(offset) or not _is_id(length) or offset < 0 or length <= 0: + if not _is_non_bool_int(offset) or not _is_non_bool_int(length) or offset < 0 or length <= 0: return None end = offset + length if end * 2 > len(units): return None try: - return offset, end, units[offset * 2:end * 2].decode("utf-16-le") + return _Span(offset, end, units[offset * 2:end * 2].decode("utf-16-le")) except UnicodeDecodeError: return None -def _address( +def _strip_bot_address( *, text: str, entities: list, bot_id: int, bot_username: str, -) -> tuple[bool, str]: +) -> _BotAddress: units = text.encode("utf-16-le") mention = "@" + bot_username.lower() - addressed, removals = False, [] + is_addressed, removals = False, [] for raw in entities: - entity = _mapping(raw) - span = _span(units, entity) + entity = _as_dict(raw) + span = _entity_span(units, entity) if span is None: continue start, end, value = span kind = entity.get("type") if kind == ENTITY_MENTION and value.lower() == mention: - addressed = True + is_addressed = True removals.append((start, end)) elif kind == ENTITY_COMMAND and value.lower().endswith(mention): - addressed = True + is_addressed = True removals.append((end - len(mention), end)) - elif kind == ENTITY_TEXT_MENTION and _mapping(entity.get("user")).get("id") == bot_id: - addressed = True + elif kind == ENTITY_TEXT_MENTION and _as_dict(entity.get("user")).get("id") == bot_id: + is_addressed = True kept, cursor = [], 0 for start, end in sorted(removals): if start < cursor: @@ -77,21 +97,21 @@ def _address( kept.append(units[cursor * 2:start * 2]) cursor = end kept.append(units[cursor * 2:]) - return addressed, b"".join(kept).decode("utf-16-le").strip() + return _BotAddress(is_addressed, b"".join(kept).decode("utf-16-le").strip()) def _attachments(message: dict) -> tuple[Attachment, ...]: found: dict[str, Attachment] = {} - photos = [item for item in _items(message.get("photo")) if isinstance(item, dict)] + photos = [item for item in _as_list(message.get("photo")) if isinstance(item, dict)] if photos and isinstance(photos[-1].get("file_id"), str) and photos[-1]["file_id"]: size = photos[-1].get("file_size") found[photos[-1]["file_id"]] = Attachment( id=photos[-1]["file_id"], media_type=PHOTO_MEDIA_TYPE, - size=size if _is_id(size) else None, + size=size if _is_non_bool_int(size) else None, ) for kind in FILE_KINDS: - media = _mapping(message.get(kind)) + media = _as_dict(message.get(kind)) file_id = media.get("file_id") if not isinstance(file_id, str) or not file_id or file_id in found: continue @@ -101,34 +121,35 @@ def _attachments(message: dict) -> tuple[Attachment, ...]: id=file_id, name=name if isinstance(name, str) and name else None, media_type=media_type if isinstance(media_type, str) and media_type else None, - size=size if _is_id(size) else None, + size=size if _is_non_bool_int(size) else None, ) return tuple(found.values()) def _button_label(message: dict, data: str) -> str: - keyboard = _mapping(message.get("reply_markup")).get("inline_keyboard") - for row in _items(keyboard): - for button in _items(row): - button = _mapping(button) + keyboard = _as_dict(message.get("reply_markup")).get("inline_keyboard") + for row in _as_list(keyboard): + for button in _as_list(row): + button = _as_dict(button) if button.get("callback_data") == data and isinstance(button.get("text"), str): return button["text"] return data -def _message( +def _build_message( *, bot_id: int, local_id: str, chat: dict, message: dict, sender: Sender, text: str, - attachments: tuple[Attachment, ...] = (), extra: dict | None = None, + attachments: tuple[Attachment, ...] = (), extra_metadata: dict | None = None, ) -> Message: thread = message.get("message_thread_id") - forum_topic = thread if message.get("is_topic_message") is True and _is_id(thread) else None - conversation_id = f"telegram:{bot_id}:{chat['id']}" + is_forum_topic = message.get("is_topic_message") is True and _is_non_bool_int(thread) + forum_topic = thread if is_forum_topic else None + conversation_id = f"{TELEGRAM_CHANNEL}:{bot_id}:{chat['id']}" return Message( - id=f"telegram:{bot_id}:{local_id}", + id=f"{TELEGRAM_CHANNEL}:{bot_id}:{local_id}", conversation_id=( conversation_id if forum_topic is None else f"{conversation_id}:{forum_topic}" ), - channel="telegram", + channel=TELEGRAM_CHANNEL, sender=sender, text=text, role="user", @@ -136,24 +157,24 @@ def _message( attachments=attachments, metadata=MappingProxyType( { - "bot_id": bot_id, - "chat_id": chat["id"], - "chat_type": chat["type"], - "message_id": message["message_id"], - "message_thread_id": forum_topic, - **(extra or {}), + METADATA_BOT_ID: bot_id, + METADATA_CHAT_ID: chat["id"], + METADATA_CHAT_TYPE: chat["type"], + METADATA_MESSAGE_ID: message["message_id"], + METADATA_THREAD_ID: forum_topic, + **(extra_metadata or {}), } ), ) -def _chat(message: dict) -> dict | None: - chat = _mapping(message.get("chat")) +def _supported_chat(message: dict) -> dict | None: + chat = _as_dict(message.get("chat")) message_id = message.get("message_id") if ( - not _is_id(chat.get("id")) + not _is_non_bool_int(chat.get("id")) or (chat.get("type") != CHAT_PRIVATE and chat.get("type") not in GROUP_CHATS) - or not _is_id(message_id) + or not _is_non_bool_int(message_id) or message_id <= 0 ): return None @@ -161,10 +182,10 @@ def _chat(message: dict) -> dict | None: def _normalize_callback(callback: dict, *, bot_id: int) -> Message | None: - message = _mapping(callback.get("message")) + message = _as_dict(callback.get("message")) data, callback_id = callback.get("data"), callback.get("id") - sender = _sender(_mapping(callback.get("from"))) - chat = _chat(message) + sender = _sender(_as_dict(callback.get("from"))) + chat = _supported_chat(message) if ( not isinstance(data, str) or not data @@ -175,42 +196,42 @@ def _normalize_callback(callback: dict, *, bot_id: int) -> Message | None: or message.get("date") == 0 ): return None - return _message( + return _build_message( bot_id=bot_id, local_id=f"callback:{callback_id}", chat=chat, message=message, sender=sender, text=_button_label(message, data), - extra={"callback_query_id": callback_id, "callback_data": data}, + extra_metadata={METADATA_CALLBACK_QUERY_ID: callback_id, METADATA_CALLBACK_DATA: data}, ) def normalize(update: object, *, bot_id: int, bot_username: str) -> Message | None: - update = _mapping(update) - if "callback_query" in update: - return _normalize_callback(_mapping(update["callback_query"]), bot_id=bot_id) - message = _mapping(update.get("message")) - sender = _sender(_mapping(message.get("from"))) - chat = _chat(message) + update = _as_dict(update) + if UPDATE_CALLBACK_QUERY in update: + return _normalize_callback(_as_dict(update[UPDATE_CALLBACK_QUERY]), bot_id=bot_id) + message = _as_dict(update.get(UPDATE_MESSAGE)) + sender = _sender(_as_dict(message.get("from"))) + chat = _supported_chat(message) if sender is None or chat is None: return None if isinstance(message.get("text"), str): - text, entities = message["text"], _items(message.get("entities")) + text, entities = message["text"], _as_list(message.get("entities")) elif isinstance(message.get("caption"), str): - text, entities = message["caption"], _items(message.get("caption_entities")) + text, entities = message["caption"], _as_list(message.get("caption_entities")) else: text, entities = "", [] - addressed, text = _address( + is_addressed, text = _strip_bot_address( text=text, entities=entities, bot_id=bot_id, bot_username=bot_username, ) - reply_to = _mapping(_mapping(message.get("reply_to_message")).get("from")) - if chat["type"] != CHAT_PRIVATE and not addressed and reply_to.get("id") != bot_id: + reply_to = _as_dict(_as_dict(message.get("reply_to_message")).get("from")) + if chat["type"] != CHAT_PRIVATE and not is_addressed and reply_to.get("id") != bot_id: return None attachments = _attachments(message) if not text and not attachments: return None - return _message( + return _build_message( bot_id=bot_id, local_id=f"{chat['id']}:{message['message_id']}", chat=chat, diff --git a/src/agentchat/channels/telegram/rendering.py b/src/agentchat/channels/telegram/rendering.py index 2a51a85..c37a4f7 100644 --- a/src/agentchat/channels/telegram/rendering.py +++ b/src/agentchat/channels/telegram/rendering.py @@ -1,5 +1,6 @@ import re import string +from typing import NamedTuple TELEGRAM_TEXT_LIMIT = 4096 TELEGRAM_CAPTION_LIMIT = 1024 @@ -39,7 +40,12 @@ def _utf16_len(text: str) -> int: return len(text) + sum(1 for char in text if ord(char) > 0xFFFF) -class _Builder: +class RenderedText(NamedTuple): + text: str + entities: list[dict] + + +class _TextBuilder: def __init__(self) -> None: self.parts: list[str] = [] self.length = 0 @@ -67,13 +73,15 @@ def _is_word_char(src: str, index: int) -> bool: return 0 <= index < len(src) and bool(_WORD_CHAR.match(src[index])) -def _render_inline(b: _Builder, src: str, active: frozenset[str] = frozenset()) -> None: +def _render_inline( + builder: _TextBuilder, src: str, active: frozenset[str] = frozenset(), +) -> None: i = 0 plain: list[str] = [] def flush() -> None: if plain: - b.add("".join(plain)) + builder.add("".join(plain)) plain.clear() while i < len(src): @@ -85,9 +93,9 @@ def flush() -> None: link = _LINK.match(src, i) if link: flush() - start = b.length - _render_inline(b, link[1], active | {_TEXT_LINK}) - b.wrap(start, _TEXT_LINK, url=link[2]) + start = builder.length + _render_inline(builder, link[1], active | {_TEXT_LINK}) + builder.wrap(start, _TEXT_LINK, url=link[2]) i = link.end() continue @@ -108,11 +116,11 @@ def flush() -> None: continue flush() if entity_type == _CODE: - b.add(inner, None if _TEXT_LINK in active else _CODE) + builder.add(inner, None if _TEXT_LINK in active else _CODE) else: - start = b.length - _render_inline(b, inner, active | {entity_type}) - b.wrap(start, entity_type) + start = builder.length + _render_inline(builder, inner, active | {entity_type}) + builder.wrap(start, entity_type) i = end + len(marker) break else: @@ -121,42 +129,42 @@ def flush() -> None: flush() -def _render_line(b: _Builder, line: str) -> None: +def _render_line(builder: _TextBuilder, line: str) -> None: if heading := _HEADING.fullmatch(line): - start = b.length - _render_inline(b, heading[1]) - b.wrap(start, _BOLD) + start = builder.length + _render_inline(builder, heading[1]) + builder.wrap(start, _BOLD) elif bullet := _BULLET.fullmatch(line): - b.add(f"{bullet[1]}โ€ข ") - _render_inline(b, bullet[2]) + builder.add(f"{bullet[1]}โ€ข ") + _render_inline(builder, bullet[2]) elif numbered := _NUMBERED.fullmatch(line): - b.add(f"{numbered[1]}{numbered[2]}. ") - _render_inline(b, numbered[3]) + builder.add(f"{numbered[1]}{numbered[2]}. ") + _render_inline(builder, numbered[3]) else: - _render_inline(b, line) + _render_inline(builder, line) -def _render_block(b: _Builder, block: list[str]) -> None: +def _render_block(builder: _TextBuilder, block: list[str]) -> None: quote_start: int | None = None for n, line in enumerate(block): quote = _QUOTE.fullmatch(line) if quote_start is not None and not quote: - b.wrap(quote_start, _BLOCKQUOTE) + builder.wrap(quote_start, _BLOCKQUOTE) quote_start = None if n: - b.add("\n") + builder.add("\n") if quote: if quote_start is None: - quote_start = b.length - _render_inline(b, quote[1]) + quote_start = builder.length + _render_inline(builder, quote[1]) else: - _render_line(b, line) + _render_line(builder, line) if quote_start is not None: - b.wrap(quote_start, _BLOCKQUOTE) + builder.wrap(quote_start, _BLOCKQUOTE) -def render(markdown: str) -> tuple[str, list[dict]]: - b = _Builder() +def render(markdown: str) -> RenderedText: + builder = _TextBuilder() lines = markdown.replace("\r\n", "\n").split("\n") i = 0 while i < len(lines): @@ -171,9 +179,9 @@ def render(markdown: str) -> tuple[str, list[dict]]: i += 1 code = "\n".join(body).rstrip("\n") if code: - if b.length: - b.add("\n\n") - b.add(code, _PRE, **({"language": language} if language else {})) + if builder.length: + builder.add("\n\n") + builder.add(code, _PRE, **({"language": language} if language else {})) continue if not lines[i].strip(): @@ -184,16 +192,15 @@ def render(markdown: str) -> tuple[str, list[dict]]: while i < len(lines) and lines[i].strip() and not _FENCE.fullmatch(lines[i]): block.append(lines[i]) i += 1 - if b.length: - b.add("\n\n") - _render_block(b, block) + if builder.length: + builder.add("\n\n") + _render_block(builder, block) - # Wrappers are appended after their contents, so equal spans list the outer entity first. order = sorted( - range(len(b.entities)), - key=lambda n: (b.entities[n]["offset"], -b.entities[n]["length"], -n), + range(len(builder.entities)), + key=lambda n: (builder.entities[n]["offset"], -builder.entities[n]["length"], -n), ) - return "".join(b.parts), [b.entities[n] for n in order] + return RenderedText("".join(builder.parts), [builder.entities[n] for n in order]) def text_payload(content: str) -> dict: diff --git a/src/agentchat/channels/telegram/transport.py b/src/agentchat/channels/telegram/transport.py index 3304d13..48e74d4 100644 --- a/src/agentchat/channels/telegram/transport.py +++ b/src/agentchat/channels/telegram/transport.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from typing import Any +from typing import Any, NamedTuple from urllib.parse import quote import aiohttp @@ -10,7 +10,22 @@ REQUEST_TIMEOUT = 30 UPLOAD_TIMEOUT = 60 DOWNLOAD_TIMEOUT = 60 -DOWNLOAD_CHUNK = 64 * 1024 +DOWNLOAD_CHUNK_BYTES = 64 * 1024 + +METHOD_GET_ME = "getMe" +METHOD_GET_UPDATES = "getUpdates" +METHOD_DELETE_WEBHOOK = "deleteWebhook" +METHOD_SET_WEBHOOK = "setWebhook" +METHOD_ANSWER_CALLBACK_QUERY = "answerCallbackQuery" +METHOD_SEND_MESSAGE = "sendMessage" +METHOD_SEND_CHAT_ACTION = "sendChatAction" +METHOD_SEND_DOCUMENT = "sendDocument" +METHOD_GET_FILE = "getFile" + + +class TelegramFile(NamedTuple): + path: str + content: bytes def _download_limit_error(max_bytes: int) -> ValueError: @@ -53,8 +68,6 @@ def _unwrap_result(method: str, status: int, body: object) -> Any: class TelegramBotClient: - """One-attempt Bot API calls; messages never carry the token-bearing URL.""" - def __init__(self, token: str, *, api_root: str = API_ROOT) -> None: self._token = token self._api_root = api_root.rstrip("/") @@ -104,8 +117,8 @@ async def upload( ) return await self._post(method, request_timeout=UPLOAD_TIMEOUT, data=form) - async def download_file(self, file_id: str, *, max_bytes: int) -> tuple[str, bytes]: - file = await self.call("getFile", {"file_id": file_id}) + async def download_file(self, file_id: str, *, max_bytes: int) -> TelegramFile: + file = await self.call(METHOD_GET_FILE, {"file_id": file_id}) file = file if isinstance(file, dict) else {} size, path = file.get("file_size"), file.get("file_path") if isinstance(size, int) and size > max_bytes: @@ -127,14 +140,14 @@ async def download_file(self, file_id: str, *, max_bytes: int) -> tuple[str, byt if response.status != 200: raise RuntimeError(f"Telegram file download failed (HTTP {response.status})") chunks, received = [], 0 - async for chunk in response.content.iter_chunked(DOWNLOAD_CHUNK): + async for chunk in response.content.iter_chunked(DOWNLOAD_CHUNK_BYTES): received += len(chunk) if received > max_bytes: raise _download_limit_error(max_bytes) chunks.append(chunk) except (aiohttp.ClientError, TimeoutError): raise TelegramError("Telegram file download failed") from None - return path, b"".join(chunks) + return TelegramFile(path, b"".join(chunks)) @property def is_closed(self) -> bool: diff --git a/tests/channels/telegram/test_polling.py b/tests/channels/telegram/test_polling.py index ab85f03..90540a1 100644 --- a/tests/channels/telegram/test_polling.py +++ b/tests/channels/telegram/test_polling.py @@ -223,8 +223,9 @@ async def handler(channel, message): def test_retry_delay_schedule(): - assert [retry_delay(n, None) for n in range(7)] == [1, 2, 4, 8, 16, 30, 30] - assert retry_delay(0, 45) == 45 + schedule = [retry_delay(attempt=n, retry_after=None) for n in range(7)] + assert schedule == [1, 2, 4, 8, 16, 30, 30] + assert retry_delay(attempt=0, retry_after=45) == 45 @pytest.mark.asyncio diff --git a/tests/support/telegram.py b/tests/support/telegram.py index 99a849d..6581905 100644 --- a/tests/support/telegram.py +++ b/tests/support/telegram.py @@ -8,7 +8,7 @@ from agentchat.channels.telegram import adapter as telegram_adapter from agentchat.channels.telegram.adapter import SECRET_HEADER from agentchat.channels.telegram.events import normalize -from agentchat.channels.telegram.transport import TelegramBotClient +from agentchat.channels.telegram.transport import TelegramBotClient, TelegramFile from tests.support.receipts import Receipt TOKEN = "42:SECRET-token_abcdef" @@ -146,7 +146,7 @@ async def upload(self, *, method, fields, file_field, filename, content): async def download_file(self, file_id, *, max_bytes): self.calls.append(("download_file", {"file_id": file_id, "max_bytes": max_bytes})) - return f"documents/{file_id}", b"content" + return TelegramFile(f"documents/{file_id}", b"content") async def close(self): self.closed = True From c7f748e6cdfee40c11320fc6cf690c7edc257435 Mon Sep 17 00:00:00 2001 From: CatalanCabbage Date: Wed, 14 Oct 2026 02:20:54 +0530 Subject: [PATCH 5/5] Telegram tests: drop duplicated cases and unused fake branches --- tests/.DS_Store | Bin 0 -> 6148 bytes tests/channels/telegram/test_outbound.py | 19 +------- tests/channels/telegram/test_rendering.py | 51 +++------------------- tests/support/telegram.py | 18 +------- 4 files changed, 9 insertions(+), 79 deletions(-) create mode 100644 tests/.DS_Store diff --git a/tests/.DS_Store b/tests/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..b6c56871ad35a8792b2aed336da77112c66f2011 GIT binary patch literal 6148 zcmeHKKT88K9Q~y&Sgk_`2SJn@oP`^lJ!}00U6t!!TA}Tg_9}=Vhm(`w;N~DWxcUji z(XZg<;O^|^`z6s{dTr4~q~uqUcS&Ay@7`UKhKNLASa*r?M3hBh%q$_B5Zulw5h-1> z0u*wL8da!9EozXTVA~4ofOX)%Ilyx_huP{=k4`XmNAuTjH{DV#==7vy@$ojD*$Ilp zYCEWck=@=pz4mvWcW1HdzhL)TF%_vYhQ(sUYfOc@_%*0a14HTBeRmQkuJPr@a}Slq z#8{nRq;3|^(6#==x)z};fToc8(#LR2S$gkcGApj_^6d8gKxvyWUR>R5rlBwW%KDb+ zNUCd4Qz{HmX1%6=V&icZH?DQ_;qFW8)Uo4=f5JC(i!$q8paBi>efFTXPoeR)Md>`r zh-*B)axe6|9vdgFY&y%(qrN}ZG!KD2nwtA&t{f1VVyRfbBk8?0@o>D?I!I+C4(3G) mWfx?6JC+x`6|W-6K%2uCKu=?&5H$$ quoted" - ) - assert entities - for e in entities: - assert e["offset"] >= 0 - assert e["length"] > 0 - assert e["offset"] + e["length"] <= len(_units(text)) // 2 - - def test_rendering_emoji_before_and_inside_entities(): text, entities = render("๐Ÿ˜€ **bold ๐Ÿ˜€ x** and `ๆ—ฅๆœฌ๐Ÿ˜€`") bold, code = _only(entities, "bold"), _only(entities, "code") @@ -91,22 +62,14 @@ def test_rendering_emoji_before_and_inside_entities(): assert code["length"] == 4 -def test_rendering_astral_emoji_then_bold(): - text, entities = render("๐Ÿ˜€ **bold**") - assert text == "๐Ÿ˜€ bold" - assert entities == [{"type": "bold", "offset": 3, "length": 4}] - - -@pytest.mark.parametrize("literal", ["edit __init__.py", "run __main__", "a __b__ c"]) -def test_rendering_double_underscore_is_not_a_delimiter(literal): - assert render(literal) == (literal, []) - - @pytest.mark.parametrize( "literal", - ["snake_case_name", "a*b*c", "get_user_data()", "x * y * z", "file_name_v2.py", "2*3*4"], + [ + "snake_case_name", "a*b*c", "get_user_data()", "x * y * z", "file_name_v2.py", "2*3*4", + "edit __init__.py", "run __main__", "a __b__ c", "[docs](./readme.md)", + ], ) -def test_rendering_intraword_and_spaced_markers_stay_literal(literal): +def test_rendering_non_delimiters_stay_literal(literal): assert render(literal) == (literal, []) @@ -155,10 +118,6 @@ def test_rendering_code_inside_link_drops_code_entity(): assert [e["type"] for e in entities] == ["text_link"] -def test_rendering_relative_link_stays_literal(): - assert render("[docs](./readme.md)") == ("[docs](./readme.md)", []) - - def test_rendering_code_containing_markers_stays_literal(): text, entities = render("`**not bold** _x_ [a](https://b.c)`") assert text == "**not bold** _x_ [a](https://b.c)" diff --git a/tests/support/telegram.py b/tests/support/telegram.py index 6581905..8b1c2f3 100644 --- a/tests/support/telegram.py +++ b/tests/support/telegram.py @@ -8,7 +8,7 @@ from agentchat.channels.telegram import adapter as telegram_adapter from agentchat.channels.telegram.adapter import SECRET_HEADER from agentchat.channels.telegram.events import normalize -from agentchat.channels.telegram.transport import TelegramBotClient, TelegramFile +from agentchat.channels.telegram.transport import TelegramBotClient from tests.support.receipts import Receipt TOKEN = "42:SECRET-token_abcdef" @@ -144,10 +144,6 @@ async def upload(self, *, method, fields, file_field, filename, content): return {"message_id": 200 + len(self.uploads), "document": {"file_id": f"file-{len(self.uploads)}"}} - async def download_file(self, file_id, *, max_bytes): - self.calls.append(("download_file", {"file_id": file_id, "max_bytes": max_bytes})) - return TelegramFile(f"documents/{file_id}", b"content") - async def close(self): self.closed = True self.stopped.set() @@ -157,10 +153,8 @@ def _ok(result): return web.json_response({"ok": True, "result": result}) -def _error(code, description, retry_after=None): +def _error(code, description): body = {"ok": False, "error_code": code, "description": description} - if retry_after: - body["parameters"] = {"retry_after": retry_after} return web.json_response(body, status=code) @@ -218,8 +212,6 @@ async def _handle(self, request): method = request.match_info["method"] payload = await self._payload(request) self.calls.append((method, payload)) - if request.match_info["token"] != self.token: - return _error(401, "Unauthorized") scripted = self.failures.get(method) if scripted: status, body = scripted.pop(0) @@ -244,13 +236,9 @@ async def _handle(self, request): if method == "getFile": file = self.files.get(payload["file_id"]) return _ok(file["info"]) if file else _error(400, "Bad Request: invalid file_id") - if method in ("sendChatAction", "answerCallbackQuery"): - return _ok(True) return _error(404, "Not Found") async def _get_updates(self, payload): - if self.webhook is not None: - return _error(409, "Conflict: can't use getUpdates method while webhook is active") offset = payload.get("offset") self.offsets.append(offset) if offset is not None: @@ -268,8 +256,6 @@ async def _get_updates(self, payload): return _ok(self.updates[:limit]) async def _download(self, request): - if request.match_info["token"] != self.token: - return web.Response(status=404) for file in self.files.values(): if file["info"].get("file_path") == request.match_info["path"] and file["content"]: return web.Response(body=file["content"])