Telegram

import asyncio import hmac import logging import re from collections import deque import httpx from bot.core import Bot from bot.providers import ProviderError # The bot token is part of every Telegram API URL, and httpx logs request URLs at INFO level. # Keep that logger quiet so the token never reaches a log file. logging.getLogger("httpx").setLevel(logging.WARNING) MESSAGE_LIMIT = 4096 # Bot API: sendMessage text is 1-4096 characters after entity parsing (as quoted by library docs) WELCOME = ("Hello! I'm an AI assistant, not a person. Ask me a question, or type /help to see what I can do " "and /privacy to see what I store.") class TelegramError(Exception): """A failed Bot API call. retry_after (seconds) is set when Telegram says to slow down.""" def __init__(self, description: str, *, error_code: int | None = None, retry_after: float | None = None) -> None: super().__init__(description) self.error_code = error_code self.retry_after = retry_after class TelegramAPI: """A small client for the Telegram Bot API: https://api.telegram.org/bot<token>/<METHOD>, JSON in and out.""" def __init__(self, token: str, *, base_url: str = "https://api.telegram.org", timeout: float = 15.0, transport: httpx.AsyncBaseTransport | None = None) -> None: self.timeout = timeout self._client = httpx.AsyncClient(base_url=f"{base_url}/bot{token}/", timeout=timeout, transport=transport) async def call(self, method: str, http_timeout: float | None = None, **params): try: response = await self._client.post(method, json=params, timeout=http_timeout or self.timeout) except httpx.HTTPError as e: # deliberately NOT chaining the original error: its text can contain the URL, and the URL contains the token raise TelegramError(f"{method}: {type(e).__name__}") from None try: body = response.json() except ValueError: raise TelegramError(f"{method}: HTTP {response.status_code} with a non-JSON body", error_code=response.status_code) from None if not body.get("ok"): retry_after = (body.get("parameters") or {}).get("retry_after") raise TelegramError(body.get("description", "unknown error"), error_code=body.get("error_code"), retry_after=float(retry_after) if retry_after is not None else None) return body["result"] async def get_me(self): return await self.call("getMe") async def get_updates(self, offset: int | None, poll_seconds: int = 30): """Long polling: Telegram holds the request open for up to poll_seconds waiting for news.""" params = {"timeout": poll_seconds, "allowed_updates": ["message"]} if offset is not None: params["offset"] = offset # one more than the highest update_id already handled: this confirms the earlier ones return await self.call("getUpdates", http_timeout=poll_seconds + 10, **params) async def send_message(self, chat_id: int, text: str, reply_to: int | None = None): params = {"chat_id": chat_id, "text": text} if reply_to is not None: params["reply_parameters"] = {"message_id": reply_to, "allow_sending_without_reply": True} return await self.call("sendMessage", **params) async def send_chat_action(self, chat_id: int, action: str = "typing"): return await self.call("sendChatAction", chat_id=chat_id, action=action) async def set_webhook(self, url: str, secret_token: str): return await self.call("setWebhook", url=url, secret_token=secret_token, allowed_updates=["message"]) async def delete_webhook(self, drop_pending_updates: bool = False): return await self.call("deleteWebhook", drop_pending_updates=drop_pending_updates) async def aclose(self) -> None: await self._client.aclose() def utf16_len(text: str) -> int: """Length in UTF-16 code units: emoji outside the basic plane count as two. A cautious way to measure a limit.""" return len(text.encode("utf-16-le")) // 2 def split_message(text: str, limit: int = MESSAGE_LIMIT) -> list[str]: """Cut a long reply into pieces that each fit the limit, preferring paragraph, line, then word boundaries.""" rest = text.strip() chunks: list[str] = [] while utf16_len(rest) > limit: fit = len(rest) while utf16_len(rest[:fit]) > limit: # the longest prefix that fits fit -= max(1, (utf16_len(rest[:fit]) - limit) // 2) window = rest[:fit] cut = fit for separator in ("\n\n", "\n", " "): at = window.rfind(separator) if at > fit // 2: # not so early that the chunk would be tiny cut = at break chunks.append(rest[:cut].rstrip()) rest = rest[cut:].lstrip() if rest: chunks.append(rest) return chunks class TelegramAdapter: """Turns Telegram updates into Bot calls and the Bot's replies into Telegram messages. The core knows nothing about Telegram; this class is the only place that does. Sessions are per chat (`telegram:<chat id>`) and users per sender (`telegram:<user id>`).""" def __init__(self, bot: Bot, api: TelegramAPI, *, username: str | None = None, bot_id: int | None = None, welcome: str = WELCOME, typing_every: float = 4.0, send_gap: float = 1.0, sleep=asyncio.sleep) -> None: self.bot, self.api = bot, api self.username, self.bot_id = username, bot_id self.welcome = welcome self.typing_every, self.send_gap, self._sleep = typing_every, send_gap, sleep self.events: list[str] = [] self._seen: deque[int] = deque(maxlen=1000) self._tasks: set[asyncio.Task] = set() async def start(self) -> None: """Ask Telegram who we are: the username is needed to recognise messages addressed to us in groups.""" me = await self.api.get_me() self.username, self.bot_id = me["username"], me["id"] # --- entry points --------------------------------------------------------------------------------- def dispatch(self, update: dict) -> None: """Handle an update in the background, so the caller (the poller, or the webhook) is not kept waiting.""" task = asyncio.create_task(self.safe_handle(update)) self._tasks.add(task) task.add_done_callback(self._tasks.discard) async def drain(self) -> None: while self._tasks: await asyncio.gather(*list(self._tasks)) async def safe_handle(self, update: dict) -> None: try: await self.handle_update(update) except Exception as e: # noqa: BLE001 - one bad update must never stop the bot self.events.append(f"update_error:{type(e).__name__}") # --- the logic ---------------------------------------------------------------------------------------- async def handle_update(self, update: dict) -> None: update_id = update.get("update_id") if update_id in self._seen: # Telegram redelivers when a webhook answers late or with an error self.events.append("duplicate_update") return if update_id is not None: self._seen.append(update_id) message = update.get("message") if not message: self.events.append("ignored:not_a_message") return sender, chat = message.get("from") or {}, message["chat"] if sender.get("is_bot") or chat["type"] == "channel": self.events.append("ignored:bot_or_channel") return private = chat["type"] == "private" text = message.get("text") if text is None: if private: await self._send(chat["id"], "I can only read text messages.", None) self.events.append("ignored:no_text") return if not private: text = self._addressed_to_us(message, text) if text is None: self.events.append("ignored:not_addressed") return elif text.startswith("/") and self.username: text = re.sub(rf"^(/\w+)@{re.escape(self.username)}\b", r"\1", text, flags=re.I) if (text.split() or [""])[0].lower() == "/start": # Telegram's "Start" button sends /start (possibly with a deep-link payload) await self._send(chat["id"], self.welcome, None if private else message["message_id"]) return reply_to = None if private else message["message_id"] typing = asyncio.create_task(self._typing(chat["id"])) try: answer = await self.bot.handle(f"telegram:{chat['id']}", text, user_id=f"telegram:{sender.get('id', chat['id'])}") except ProviderError as e: self.events.append(f"provider_error:{type(e).__name__}") answer = self.bot.error_reply finally: typing.cancel() await asyncio.gather(typing, return_exceptions=True) for number, piece in enumerate(split_message(answer)): if number: await self._sleep(self.send_gap) # Telegram asks bots to stay under about one message a second per chat await self._send(chat["id"], piece, reply_to if number == 0 else None) def _addressed_to_us(self, message: dict, text: str) -> str | None: """In a group, answer only commands aimed at us, mentions of us, and replies to us. Returns the cleaned text or None.""" if not self.username: return None tag = f"@{self.username}" replied_to_us = ((message.get("reply_to_message") or {}).get("from") or {}).get("id") == self.bot_id mentioned = tag.lower() in text.lower() if not (mentioned or replied_to_us): return None cleaned = re.sub(re.escape(tag), "", text, flags=re.I).strip() return cleaned or None async def _typing(self, chat_id: int) -> None: """Show 'typing...' while the reply is being prepared. Telegram clears it after about five seconds, so repeat it.""" while True: try: await self.api.send_chat_action(chat_id, "typing") except TelegramError: pass await self._sleep(self.typing_every) async def _send(self, chat_id: int, text: str, reply_to: int | None) -> None: for attempt in range(1, 4): try: await self.api.send_message(chat_id, text, reply_to) return except TelegramError as e: if e.retry_after is None or attempt == 3: self.events.append(f"send_failed:{e.error_code}") return await self._sleep(e.retry_after) # flood control: wait exactly as long as Telegram says async def run_polling(adapter: TelegramAdapter, api: TelegramAPI, *, poll_seconds: int = 30, stop: asyncio.Event | None = None, sleep=asyncio.sleep, max_backoff: float = 30.0) -> None: """Long polling: ask Telegram for updates in a loop. Outbound connections only, so it works behind a firewall.""" await adapter.start() offset: int | None = None failures = 0 while not (stop and stop.is_set()): try: updates = await api.get_updates(offset, poll_seconds) except TelegramError as e: failures += 1 adapter.events.append(f"poll_error:{e.error_code}") await sleep(e.retry_after or min(max_backoff, 2 ** failures)) continue failures = 0 for update in updates: offset = update["update_id"] + 1 # confirms this update (and all before it) on the next call adapter.dispatch(update) await adapter.drain() def add_webhook_route(app, adapter: TelegramAdapter, *, secret_token: str, path: str = "/telegram/webhook") -> None: """Webhook mode: Telegram POSTs updates to us. The secret token header proves the request is from Telegram.""" from fastapi import Request from fastapi.responses import JSONResponse @app.post(path) async def webhook(request: Request): given = request.headers.get("x-telegram-bot-api-secret-token", "") if not hmac.compare_digest(given, secret_token): return JSONResponse({"detail": "forbidden"}, status_code=403) adapter.dispatch(await request.json()) return {"ok": True} # answer at once; the reply is sent in the background

"""Run the bot on Telegram by long polling. set TELEGRAM_BOT_TOKEN=<the token BotFather gave you> set CHATBOT_PROVIDER=fake (or claude / grok / openai / ollama: see the provider table in chapter 2) py run_telegram.py """ import asyncio import os import pathlib import sys from bot.channels.telegram import TelegramAPI, TelegramAdapter, run_polling from bot.core import Bot from bot.limits import Limits from bot.moderation import KeywordModerator, SafetyPolicy from bot.persona import load_persona from bot.providers import ProviderConfig, make_provider from bot.reliability import BreakerProvider, FallbackProvider, RetryingProvider from bot.store import Store CONFIGS = { "fake": ProviderConfig("fake"), "claude": ProviderConfig("claude", model=os.environ.get("CHATBOT_MODEL", ""), api_key_env="ANTHROPIC_API_KEY"), "grok": ProviderConfig("openai-compatible", model=os.environ.get("CHATBOT_MODEL", ""), base_url="https://api.x.ai/v1", api_key_env="XAI_API_KEY"), "openai": ProviderConfig("openai-compatible", model=os.environ.get("CHATBOT_MODEL", ""), base_url="https://api.openai.com/v1", api_key_env="OPENAI_API_KEY"), "ollama": ProviderConfig("openai-compatible", model=os.environ.get("CHATBOT_MODEL", ""), base_url="http://localhost:11434/v1"), } async def main() -> None: token = os.environ.get("TELEGRAM_BOT_TOKEN") if not token: sys.exit("Set TELEGRAM_BOT_TOKEN to the token BotFather gave you (it looks like 123456:ABC-DEF...).") choice = os.environ.get("CHATBOT_PROVIDER", "fake") provider = BreakerProvider(RetryingProvider(make_provider(CONFIGS[choice]))) persona = load_persona(pathlib.Path("persona.toml").read_text(encoding="utf8")) bot = Bot(provider, persona=persona, store=Store("telegram_bot.db"), moderator=KeywordModerator({"harassment": ("you idiot",)}), safety=SafetyPolicy(), limits=Limits(burst=5, per_minute=20, user_daily_tokens=50_000)) api = TelegramAPI(token) adapter = TelegramAdapter(bot, api) stop = asyncio.Event() try: await run_polling(adapter, api, stop=stop) except KeyboardInterrupt: stop.set() finally: await api.aclose() if __name__ == "__main__": asyncio.run(main())