Wie man einen Telegram-Bot baut, der nur dir antwortet, mit Long Polling in Python

Ein Telegram-Bot ist eine bequeme Fernbedienung für ein Skript oder einen KI-Agenten auf dem eigenen Rechner. Er funktioniert auf jedem Handy, braucht keine eigene App und keinen öffentlichen Server. Der Haken: Ein Bot ist öffentlich. Jeder kann seinen Benutzernamen finden und ihm schreiben, und das Token beweist nur, dass dein Skript der Bot ist; wer schreibt, sagt es nicht. Die osFoundry-Desktop-App hat für ihre Agenten eine solche Brücke, geschrieben in Rust (telegram_bridge.rs). Unten steht dasselbe Design in etwa 70 Zeilen Python mit reiner Standardbibliothek, getestet gegen einen lokalen Ersatz der Bot API, dazu der Ein-Zeilen-Fehler, der aus einem ruhigen Bot 183.389 Anfragen pro Minute machte.

Long Polling statt Webhook

Telegram liefert Nachrichten auf zwei Wegen aus. Ein Webhook braucht eine öffentliche HTTPS-URL, die Telegram aufrufen kann. getUpdates braucht nichts, was von außen erreichbar ist: Das Skript fragt nach neuen Nachrichten, und mit timeout=25 hält Telegram die Anfrage bis zu 25 Sekunden offen und antwortet, sobald eine eintrifft. Auf einem Laptop ist das die richtige Wahl. Es gibt keinen Port zu öffnen, keinen Tunnel und kein Zertifikat.

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

Das HTTP-Timeout (40 s) muss länger sein als die Haltezeit des Long Polls (25 s). Ist es kürzer, endet jeder ruhige Zyklus in einem clientseitigen Timeout, das wie ein Ausfall aussieht. Die App speichert das Token im Schlüsselbund des Betriebssystems, nicht in einer Datei. In Python macht keyring.get_password dasselbe, und für einen Test genügt eine Umgebungsvariable.

In 60 Sekunden Leerlauf machte die Long-Poll-Schleife 3 Anfragen, ein Short Poll im Sekundentakt 60. Eine Nachricht, die ankam, während eine Anfrage offen gehalten wurde, bekam im lokalen Test 1,5 bis 6,3 ms später ihre Antwort. Im echten Netz kommt die Laufzeit zu Telegram noch hinzu.

Pairing: Wer darf sprechen

Der Bot muss wissen, welcher Person er dient, bevor er irgendetwas tut. Die App löst das mit einem Einmalcode. Du klickst am Rechner auf "Pair a phone", er zeigt acht zufällige Zeichen, und du schickst /pair CODE vom Handy an den Bot. Wer innerhalb von 10 Minuten einen gültigen Code sendet, ist der Besitzer.

pending = {} # code -> created_at paired = {} # sender id -> chat id  def begin_pairing(): code = secrets.token_hex(4).upper() pending[code] = time.time() return code 

Der Code stammt aus secrets, nicht aus etwas Erratbarem wie der Uhrzeit oder dem Bot-Namen, denn der Code ist das Einzige, was einen Absender autorisiert. Er wird außerdem beim Benutzen entfernt und funktioniert deshalb nur einmal.

Gespeichert wird die numerische from.id, nicht der @username und nicht der Anzeigename. Beide lassen sich ändern, und ein frei gewordener Benutzername kann von jemand anderem registriert werden. Die Chat-ID wird daneben gespeichert, aus einem Grund, der weiter unten kommt.

Entscheiden, was mit einer Nachricht passiert

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

Drei Regeln tragen die Sicherheit, und jede geht auf einen konkreten Fehler zurück, der im Quelltext der App vermerkt ist.

Unbekannte Absender bekommen Stille. Eine Antwort wie "you are not paired" verrät einem Fremden, dass der Bot mit dem Rechner einer echten Person verbunden ist, und lädt ihn ein, es weiter zu versuchen. Keine Antwort sieht aus wie ein Bot, den niemand betreibt.

Pairing funktioniert nur in einem privaten Chat. Ein Bot, der zu einer Gruppe hinzugefügt wird, bekommt die Nachrichten der Gruppe; Pairing dort würde jedem Mitglied erlauben, dem Agenten Anweisungen zu geben.

Eine gepairte Person bekommt nur in dem Chat eine Antwort, aus dem sie gepairt hat. Die Kommentare der App beschreiben eine frühere Version, die den Absender prüfte, aber in den Chat antwortete, aus dem die Nachricht kam. Ein gepairter Nutzer, der den Bot einer Gruppe hinzufügte, hatte dann einen Agenten, der in dieser Gruppe antwortete, wo jeder Werkzeuge auf dem Rechner des Besitzers steuern konnte. Autorisiert wird nach Absender, und die Antwort geht an einen Chat, also müssen beide zusammen geprüft werden.

/unpair wird vor dieser Chat-Prüfung behandelt, damit sich der Zugriff vom Handy aus entziehen lässt, wo man ist, wenn man es braucht.

Die Schleife und der 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 ist die Art, wie das Skript Telegram mitteilt, welche Updates es gesehen hat. Alles darunter ist bestätigt und wird verworfen, alles auf oder über diesem Wert wird erneut gesendet. Der Kommentar in dieser Zeile ist das Wichtige. Es ist verlockend, den Offset erst zu verschieben, nachdem man etwas mit einer Nachricht getan hat, denn das fühlt sich nach "verarbeitet" an. Aber eine Nachricht von einem Fremden wird absichtlich ignoriert, also nie bestätigt, also kommt sie beim nächsten Aufruf sofort wieder, weil ein Update ansteht und Telegram keinen Grund hat, die Anfrage offen zu halten.

Das Backoff ist für den anderen Fehler: Ein widerrufenes oder falsch eingetipptes Token scheitert bei jeder Anfrage, und mit voller Geschwindigkeit zu wiederholen heißt, alle paar Sekunden ein 401 für immer. Die Treppe aus 5, 15, 60 und dann 300 Sekunden ist die, die die App verwendet.

Was die Tests zeigten

Der lokale Ersatz implementiert getUpdates (Offset-Bestätigung und Timeout-Haltezeit) und sendMessage und protokolliert jeden Aufruf. In jedem Szenario wurde eine Nachricht eingespeist und dann gelesen, was der Bot zurückschickte.

Dann liefen vier Schleifen 60 Sekunden lang, jede gegen ihren eigenen Ersatz:

Auf dem Ersatz, der vom selben Rechner antwortet, lief der Fehler mit etwa 3.000 Anfragen pro Sekunde. Gegen die echte API liefe die Schleife so schnell, wie die Netzwerklaufzeit es zulässt, wobei bei jedem Aufruf dasselbe "hi" des Fremden erneut zugestellt würde und nie etwas Neueres ankäme. Eine einzige Nachricht von jemandem, dem du nie antworten wirst, reicht, um das auszulösen.

Was fehlt

Eine echte Brücke muss außerdem Antworten unter Telegrams Grenze von 4.096 Zeichen halten (die App schneidet bei 3.900 ab, an einer Zeichengrenze, weil ein Byte-Schnitt mitten in einem japanischen Zeichen landen kann) und Antworten als reinen Text senden. Bei Markdown lehnt Telegram die ganze Nachricht ab, wenn ein Sternchen oder Unterstrich nicht ausgeglichen ist, und ein Agent, der über snake_case-Code schreibt, erzeugt genau das. Am wichtigsten: Der ankommende Text stammt von dem, der das Handy hält, und sollte deshalb als nicht vertrauenswürdige Eingabe beim Agenten ankommen, genau wie eine E-Mail. Pairing entscheidet, wer mit dem Agenten sprechen darf; es macht nicht sicher, was gesagt wird.

Die osFoundry-Desktop-App liefert diese Brücke für ihre KI-Agenten mit: Device → Background für das Token und "Pair a phone" im Agents-Panel eines Kanals. Sie ist auf der osFoundry-Downloadseite kostenlos.