Comment créer un bot Telegram qui ne répond qu'à vous, avec le long polling en Python

Un bot Telegram est une télécommande pratique pour un script ou un agent IA sur votre propre machine. Il fonctionne sur tous les téléphones, n'exige aucune application de votre part et aucun serveur public. Le piège : un bot est public. N'importe qui peut trouver son nom d'utilisateur et lui écrire, et le jeton prouve seulement que votre script est le bot ; il ne dit rien de celui qui écrit. L'application de bureau osFoundry a un pont de ce genre pour ses agents, écrit en Rust (telegram_bridge.rs). Voici la même conception en environ 70 lignes de Python avec la seule bibliothèque standard, testée contre un faux local de l'API Bot, ainsi que l'erreur d'une ligne qui a transformé un bot tranquille en 183 389 requêtes par minute.

Long polling plutôt qu'un webhook

Telegram livre les messages de deux façons. Un webhook exige une URL HTTPS publique que Telegram puisse appeler. getUpdates n'expose rien : le script demande les nouveaux messages et, avec timeout=25, Telegram garde la requête ouverte jusqu'à 25 secondes et répond dès qu'un message arrive. Sur un portable, c'est le bon choix. Pas de port à ouvrir, pas de tunnel, pas de certificat.

import json, os, secrets, time, urllib.parse, urllib.request

API = os.environ.get("TG_API", "https://api.telegram.org")
TOKEN = os.environ["TG_TOKEN"]
LONG_POLL = 25
PAIR_TTL = 600

def call(method, params=None, body=None, timeout=40):
    url = f"{API}/bot{TOKEN}/{method}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data, {"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.load(r)

def send(chat_id, text):
    call("sendMessage", body={"chat_id": chat_id, "text": text[:3900]})

Le timeout HTTP (40 s) doit être plus long que l'attente du long poll (25 s). S'il est plus court, chaque cycle calme se termine par un timeout côté client qui ressemble à une panne. L'application garde le jeton dans le trousseau du système, pas dans un fichier. En Python, keyring.get_password fait la même chose, et une variable d'environnement suffit pour un test.

Sur 60 secondes d'inactivité, la boucle de long poll a fait 3 requêtes et un short poll toutes les secondes en a fait 60. Un message arrivé pendant qu'une requête était maintenue a reçu sa réponse 1,5 à 6,3 ms plus tard dans le test local. Sur un vrai réseau, l'aller-retour vers Telegram s'ajoute.

L'appairage : qui a le droit de parler

Le bot doit savoir quelle personne il sert avant de faire quoi que ce soit. L'application le fait avec un code à usage unique. Vous cliquez sur « Pair a phone » sur la machine, elle affiche huit caractères aléatoires, et vous envoyez /pair CODE au bot depuis votre téléphone. Celui qui envoie un code valide dans les 10 minutes est le propriétaire.

pending = {}   # code -> created_at
paired = {}    # sender id -> chat id

def begin_pairing():
    code = secrets.token_hex(4).upper()
    pending[code] = time.time()
    return code

Le code vient de secrets, et non de quelque chose de devinable comme l'heure ou le nom du bot, car le code est la seule chose qui autorise un expéditeur. Il est aussi supprimé à l'usage, donc il ne sert qu'une fois.

Ce qui est stocké, c'est le from.id numérique, pas le @username ni le nom affiché. Les deux peuvent être modifiés, et un nom d'utilisateur libéré peut être enregistré par quelqu'un d'autre. L'identifiant du chat est stocké à côté, pour une raison qui apparaît plus bas.

Décider quoi faire d'un message

def handle(msg, run_agent):
    text = (msg.get("text") or "").strip()
    sender = msg.get("from", {}).get("id")
    chat = msg["chat"]
    if not text or sender is None:
        return
    words = text.split()
    if words[0] == "/pair" or words[0].startswith("/pair@"):
        if chat.get("type") != "private":
            return send(chat["id"], "Pair with me in a direct message, not in a group.")
        code = words[1].upper() if len(words) > 1 else ""
        created = pending.pop(code, None)
        if created is None or time.time() - created > PAIR_TTL:
            return send(chat["id"], "That code is not valid or has expired.")
        paired[sender] = chat["id"]
        return send(chat["id"], "Paired. Message me and the agent will answer.")
    if text.lower() == "/unpair":
        gone = paired.pop(sender, None) is not None
        return send(chat["id"], "Unpaired." if gone else "You were not paired.")
    if paired.get(sender) != chat["id"] or chat.get("type") != "private":
        return  # silence
    send(chat["id"], run_agent(text))

Trois règles portent la sécurité, et chacune vient d'une défaillance précise notée dans le code source de l'application.

Les expéditeurs inconnus reçoivent le silence. Une réponse comme « you are not paired » apprend à un inconnu que le bot est relié à la machine d'une vraie personne et l'invite à continuer d'essayer. Pas de réponse ressemble à un bot que personne n'exécute.

L'appairage ne fonctionne que dans une conversation privée. Un bot ajouté à un groupe reçoit les messages du groupe, donc appairer là permettrait à chaque membre de donner des instructions à l'agent.

Une personne appairée n'obtient de réponse que dans le chat depuis lequel elle s'est appairée. Les commentaires de l'application décrivent une version antérieure qui vérifiait l'expéditeur mais répondait au chat d'où venait le message. Un utilisateur appairé qui ajoutait le bot à un groupe avait alors un agent qui répondait dans ce groupe, où tout le monde pouvait piloter des outils sur la machine du propriétaire. L'autorisation se fait par expéditeur et la réponse va à un chat, donc les deux doivent être vérifiés ensemble.

/unpair est traité avant cette vérification du chat, pour que l'accès puisse être retiré depuis le téléphone, là où vous êtes quand vous en avez besoin.

La boucle et l'offset

def poll(run_agent):
    offset, failures = 0, 0
    while True:
        try:
            res = call("getUpdates", {"timeout": LONG_POLL, "offset": offset,
                                      "allowed_updates": '["message"]'})
            failures = 0
        except Exception:
            failures += 1
            time.sleep({1: 5, 2: 15, 3: 60}.get(failures, 300))
            continue
        for u in res["result"]:
            offset = max(offset, u["update_id"] + 1)   # even for messages we ignore
            if "message" in u:
                handle(u["message"], run_agent)

offset est la façon dont le script indique à Telegram quelles mises à jour il a vues. Tout ce qui est en dessous est confirmé et supprimé, et tout ce qui est à cette valeur ou au-dessus est renvoyé. Le commentaire sur cette ligne est le point important. Il est tentant de ne déplacer l'offset qu'après avoir fait quelque chose du message, car cela ressemble à « traité ». Mais le message d'un inconnu est volontairement ignoré, donc jamais confirmé, donc il revient à l'appel suivant, immédiatement, car une mise à jour est en attente et Telegram n'a aucune raison de garder la requête ouverte.

Le backoff vise l'autre défaillance : un jeton révoqué ou mal saisi échoue à chaque requête, et réessayer à pleine vitesse revient à un 401 toutes les quelques secondes, pour toujours. L'échelle de 5, 15, 60 puis 300 secondes est celle qu'utilise l'application.

Ce que les tests ont montré

Le faux local implémente getUpdates (confirmation par offset et attente du timeout) et sendMessage, et enregistre chaque appel. Chaque scénario a injecté un message puis lu ce que le bot renvoyait.

Message	Réponse
inconnu : « hi »	aucune
inconnu : /pair 12345678	That code is not valid or has expired.
moi, dans un groupe : /pair CODE	Pair with me in a direct message, not in a group.
moi, en DM : /pair code (minuscules)	Paired.
inconnu, réutilisant mon code	That code is not valid or has expired.
moi, en DM : une question	la réponse de l'agent
moi, dans le groupe : une question	aucune
moi : /unpair, puis une question	Unpaired., puis aucune

Ensuite, quatre boucles ont tourné pendant 60 secondes, chacune contre son propre faux :

Boucle	appels à getUpdates en 60 s
long poll, aucun message	3
short poll chaque seconde, aucun message	60
un message d'inconnu, offset avancé pour chaque mise à jour	4
un message d'inconnu, offset avancé seulement pour les messages traités	183 389

Sur le faux, qui répond depuis la même machine, le bug tournait à environ 3 000 requêtes par seconde. Contre la vraie API, la boucle irait aussi vite que l'aller-retour réseau le permet, avec le même « hi » de l'inconnu renvoyé à chaque appel et rien de plus récent qui n'arrive. Un seul message de quelqu'un à qui vous ne répondrez jamais suffit à tout déclencher.

Ce qui est laissé de côté

Un vrai pont doit aussi limiter les réponses sous la limite de 4 096 caractères de Telegram (l'application coupe à 3 900, à une frontière de caractère, car une coupe par octets peut tomber au milieu d'un caractère japonais), et envoyer les réponses en texte brut. Avec Markdown, Telegram rejette tout le message si un astérisque ou un tiret bas n'est pas équilibré, et un agent qui parle de code en snake_case produit exactement cela. Surtout, le texte qui arrive est écrit par celui qui tient le téléphone, il doit donc parvenir à l'agent comme une entrée non fiable, au même titre qu'un e-mail. L'appairage décide qui peut parler à l'agent ; il ne rend pas sûr ce qui est dit.

L'application de bureau osFoundry fournit ce pont pour ses agents IA : Device → Background pour le jeton, et « Pair a phone » dans le panneau Agents d'un canal. Elle est gratuite sur la page de téléchargement d'osFoundry.