Come creare un bot Telegram che risponde solo a te, con il long polling in Python

Un bot Telegram è un comodo telecomando per uno script o un agente IA sulla tua macchina. Funziona su qualsiasi telefono, non richiede una tua app e non richiede un server pubblico. Il problema è che un bot è pubblico. Chiunque può trovarne lo username e scrivergli, e il token dimostra soltanto che il tuo script è il bot; non dice nulla su chi sta scrivendo. L'app desktop osFoundry ha un ponte del genere per i suoi agenti, scritto in Rust (telegram_bridge.rs). Qui sotto trovi lo stesso schema in circa 70 righe di Python con la sola libreria standard, testato contro un sostituto locale della Bot API, insieme all'errore di una riga che ha trasformato un bot tranquillo in 183.389 richieste al minuto.

Long polling invece di un webhook

Telegram consegna i messaggi in due modi. Un webhook richiede un URL HTTPS pubblico che Telegram possa chiamare. getUpdates non richiede di esporre nulla: lo script chiede i nuovi messaggi e, con timeout=25, Telegram tiene aperta la richiesta fino a 25 secondi e risponde appena ne arriva uno. Su un portatile è la scelta giusta. Nessuna porta da aprire, nessun tunnel, nessun certificato.

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]})

Il timeout HTTP (40 s) deve essere più lungo dell'attesa del long poll (25 s). Se è più corto, ogni ciclo tranquillo finisce con un timeout lato client che sembra un'interruzione. L'app conserva il token nel portachiavi del sistema operativo, non in un file. Da Python, keyring.get_password fa lo stesso, e per un test va bene una variabile d'ambiente.

In 60 secondi di inattività, il ciclo di long poll ha fatto 3 richieste e uno short poll ogni secondo ne ha fatte 60. Un messaggio arrivato mentre una richiesta era tenuta aperta ha ricevuto la risposta da 1,5 a 6,3 ms dopo nel test locale. Su una rete reale si aggiunge il tragitto di andata e ritorno verso Telegram.

Pairing: chi può parlare

Il bot deve sapere quale persona serve prima di fare qualsiasi cosa. L'app lo fa con un codice monouso. Clicchi "Pair a phone" sulla macchina, che mostra otto caratteri casuali, e dal telefono invii /pair CODE al bot. Chi invia un codice valido entro 10 minuti è il proprietario.

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

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

Il codice viene da secrets, non da qualcosa di indovinabile come l'ora o il nome del bot, perché il codice è l'unica cosa che autorizza un mittente. Viene anche rimosso quando è usato, quindi funziona una sola volta.

Quello che viene salvato è il from.id numerico, non l'@username né il nome visualizzato. Entrambi possono essere cambiati, e uno username liberato può essere registrato da qualcun altro. Accanto viene salvato l'id della chat, per un motivo che compare più sotto.

Decidere cosa fare con un messaggio

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))

Tre regole reggono la sicurezza, e ciascuna nasce da un errore specifico annotato nel sorgente dell'app.

I mittenti sconosciuti ricevono silenzio. Una risposta come "you are not paired" dice a un estraneo che il bot è collegato alla macchina di una persona reale e lo invita a continuare a provare. Nessuna risposta sembra un bot che nessuno esegue.

Il pairing funziona solo in una chat privata. Un bot aggiunto a un gruppo riceve i messaggi del gruppo, quindi fare il pairing lì permetterebbe a ogni membro di dare istruzioni all'agente.

A una persona associata si risponde solo nella chat da cui ha fatto il pairing. I commenti dell'app descrivono una versione precedente che controllava il mittente ma rispondeva alla chat da cui arrivava il messaggio. Un utente associato che aggiungeva il bot a un gruppo si ritrovava con un agente che rispondeva in quel gruppo, dove tutti potevano usare strumenti sulla macchina del proprietario. L'autorizzazione è per mittente e la risposta va a una chat, quindi i due controlli vanno fatti insieme.

/unpair viene gestito prima di quel controllo sulla chat, così l'accesso si può togliere dal telefono, che è dove sei quando ti serve.

Il ciclo e 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 è il modo in cui lo script dice a Telegram quali aggiornamenti ha visto. Tutto ciò che sta sotto è confermato e scartato, e tutto ciò che è pari o sopra viene inviato di nuovo. Il commento su quella riga è la parte importante. È allettante spostare l'offset solo dopo aver fatto qualcosa con il messaggio, perché sembra "elaborato". Ma il messaggio di un estraneo viene ignorato di proposito, quindi non viene mai confermato, quindi ritorna alla chiamata successiva, subito, perché c'è un aggiornamento in sospeso e Telegram non ha motivo di tenere la richiesta aperta.

Il backoff serve per l'altro guasto: un token revocato o digitato male fallisce a ogni richiesta, e riprovare a piena velocità significa un 401 ogni pochi secondi, per sempre. La scala di 5, 15, 60 e poi 300 secondi è quella che usa l'app.

Cosa hanno mostrato i test

Il sostituto locale implementa getUpdates (conferma tramite offset e attesa del timeout) e sendMessage, e registra ogni chiamata. Ogni scenario ha iniettato un messaggio e poi letto ciò che il bot rispondeva.

Messaggio	Risposta
estraneo: "hi"	nessuna
estraneo: /pair 12345678	That code is not valid or has expired.
io, in un gruppo: /pair CODE	Pair with me in a direct message, not in a group.
io, in un DM: /pair code (minuscolo)	Paired.
estraneo, che riusa il mio codice	That code is not valid or has expired.
io, nel DM: una domanda	la risposta dell'agente
io, nel gruppo: una domanda	nessuna
io: /unpair, poi una domanda	Unpaired., poi nessuna

Poi quattro cicli sono stati eseguiti per 60 secondi, ciascuno contro il proprio sostituto:

Ciclo	chiamate a getUpdates in 60 s
long poll, nessun messaggio	3
short poll ogni secondo, nessun messaggio	60
un messaggio di un estraneo, offset avanzato per ogni aggiornamento	4
un messaggio di un estraneo, offset avanzato solo per i messaggi gestiti	183.389

Sul sostituto, che risponde dalla stessa macchina, il bug girava a circa 3.000 richieste al secondo. Contro la vera API, il ciclo andrebbe veloce quanto permette il tragitto di rete, con lo stesso "hi" dell'estraneo riconsegnato a ogni chiamata e nulla di più recente in arrivo. Basta un messaggio di qualcuno a cui non risponderai mai per far partire tutto.

Cosa è stato lasciato fuori

Un vero ponte deve anche limitare le risposte sotto il limite di 4.096 caratteri di Telegram (l'app taglia a 3.900, su un confine di carattere, perché un taglio a byte può cadere nel mezzo di un carattere giapponese) e inviare le risposte come testo semplice. Con il Markdown, Telegram rifiuta l'intero messaggio se un asterisco o un trattino basso non è bilanciato, e un agente che scrive di codice in snake_case produce esattamente questo. Soprattutto, il testo in arrivo è scritto da chi ha il telefono in mano, quindi deve arrivare all'agente come input non attendibile, come un'email. Il pairing decide chi può parlare con l'agente; non rende sicuro ciò che viene detto.

L'app desktop osFoundry include questo ponte per i suoi agenti IA: Device → Background per il token e "Pair a phone" nel pannello Agents di un canale. È gratuita dalla pagina di download di osFoundry.