Python में long polling से ऐसा Telegram बॉट कैसे बनाएँ जो सिर्फ़ आपको जवाब दे

Telegram बॉट आपकी अपनी मशीन पर चल रही स्क्रिप्ट या AI एजेंट के लिए एक सुविधाजनक रिमोट कंट्रोल है। यह हर फ़ोन पर चलता है, आपको अपना ऐप नहीं बनाना पड़ता, और किसी पब्लिक सर्वर की ज़रूरत नहीं होती। दिक्कत यह है कि बॉट पब्लिक होता है। कोई भी उसका यूज़रनेम ढूँढकर उसे मैसेज भेज सकता है, और टोकन सिर्फ़ यह साबित करता है कि आपकी स्क्रिप्ट ही वह बॉट है; यह नहीं बताता कि लिख कौन रहा है। osFoundry डेस्कटॉप ऐप में अपने एजेंटों के लिए ऐसा ही एक ब्रिज है, जो Rust में लिखा है (telegram_bridge.rs)। नीचे वही डिज़ाइन लगभग 70 लाइन की स्टैंडर्ड-लाइब्रेरी Python में है, Bot API के एक लोकल नकली सर्वर के सामने टेस्ट किया हुआ, और साथ में वह एक लाइन की गलती जिसने शांत बॉट को 183,389 रिक्वेस्ट प्रति मिनट पर पहुँचा दिया।

Webhook की जगह long polling

Telegram मैसेज दो तरीकों से पहुँचाता है। Webhook के लिए एक पब्लिक HTTPS URL चाहिए जिसे Telegram कॉल कर सके। getUpdates को कुछ भी बाहर खोलने की ज़रूरत नहीं: स्क्रिप्ट नए मैसेज माँगती है, और timeout=25 के साथ Telegram रिक्वेस्ट को 25 सेकंड तक खुला रखता है और मैसेज आते ही जवाब देता है। लैपटॉप पर यही सही चुनाव है। कोई पोर्ट नहीं खोलना, कोई टनल नहीं, कोई सर्टिफ़िकेट नहीं।

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

HTTP timeout (40 s) long-poll के रोकने के समय (25 s) से लंबा होना चाहिए। अगर यह छोटा है, तो हर शांत चक्र क्लाइंट-साइड timeout पर खत्म होता है, जो आउटेज जैसा दिखता है। ऐप टोकन को OS कीचेन में रखता है, फ़ाइल में नहीं। Python से keyring.get_password वही काम करता है, और टेस्ट के लिए एनवायरनमेंट वेरिएबल ठीक है।

60 सेकंड की निष्क्रियता में long-poll लूप ने 3 रिक्वेस्ट कीं और हर सेकंड वाले short poll ने 60। रिक्वेस्ट रुकी होने के दौरान जो मैसेज आया, उसे लोकल टेस्ट में 1.5 से 6.3 ms बाद जवाब मिल गया। असली नेटवर्क पर Telegram तक का राउंड ट्रिप इसके ऊपर जुड़ता है।

Pairing: कौन बात कर सकता है

कुछ भी करने से पहले बॉट को पता होना चाहिए कि वह किस व्यक्ति की सेवा कर रहा है। ऐप यह एक बार वाले कोड से करता है। आप मशीन पर "Pair a phone" पर क्लिक करते हैं, वह आठ रैंडम अक्षर दिखाती है, और आप फ़ोन से बॉट को /pair CODE भेजते हैं। 10 मिनट के अंदर जो भी सही कोड भेजे, वही मालिक है।

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

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

कोड secrets से बनता है, समय या बॉट के नाम जैसी अनुमान लगाने लायक चीज़ से नहीं, क्योंकि कोड ही भेजने वाले को अधिकृत करने वाली अकेली चीज़ है। इस्तेमाल होने पर इसे हटा भी दिया जाता है, इसलिए यह सिर्फ़ एक बार चलता है।

जो रखा जाता है वह संख्या वाला from.id है, @username या डिस्प्ले नाम नहीं। इन दोनों को बदला जा सकता है, और छूटा हुआ यूज़रनेम कोई और रजिस्टर कर सकता है। चैट id उसके बगल में रखी जाती है, जिसकी वजह नीचे आती है।

मैसेज के साथ क्या करना है, यह तय करना

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

सुरक्षा तीन नियमों पर टिकी है, और हर नियम ऐप के सोर्स में दर्ज एक ख़ास गड़बड़ी से निकला है।

अनजान भेजने वालों को चुप्पी मिलती है। "you are not paired" जैसा जवाब किसी अजनबी को बता देता है कि बॉट किसी असली व्यक्ति की मशीन से जुड़ा है और उसे कोशिश करते रहने का न्योता देता है। जवाब न देना ऐसे बॉट जैसा दिखता है जिसे कोई चला ही नहीं रहा।

Pairing सिर्फ़ प्राइवेट चैट में चलती है। ग्रुप में जोड़े गए बॉट को ग्रुप के मैसेज मिलते हैं, इसलिए वहाँ pairing से ग्रुप का हर सदस्य एजेंट को निर्देश दे सकता।

Pair किए गए व्यक्ति को सिर्फ़ उसी चैट में जवाब मिलता है जिससे उसने pair किया था। ऐप के कमेंट एक पुराने वर्ज़न का ज़िक्र करते हैं जो भेजने वाले को जाँचता था पर मैसेज जिस भी चैट से आया हो वहीं जवाब देता था। Pair किया हुआ यूज़र जब बॉट को ग्रुप में जोड़ता था, तब एजेंट उस ग्रुप में जवाब दे रहा होता, जहाँ हर कोई मालिक की मशीन पर टूल चला सकता था। अधिकृत करना भेजने वाले के हिसाब से होता है और जवाब एक चैट में जाता है, इसलिए दोनों को साथ में जाँचना ज़रूरी है।

/unpair को उस चैट जाँच से पहले संभाला जाता है, ताकि एक्सेस फ़ोन से हटाया जा सके, जहाँ आप तब होते हैं जब इसकी ज़रूरत पड़ती है।

लूप और 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 वह तरीका है जिससे स्क्रिप्ट Telegram को बताती है कि उसने कौन से अपडेट देख लिए हैं। इससे नीचे का सब कुछ कन्फ़र्म होकर हटा दिया जाता है, और इस मान पर या इससे ऊपर का सब कुछ दोबारा भेजा जाता है। उस लाइन का कमेंट ही असली बात है। offset को तभी आगे बढ़ाने का मन करता है जब मैसेज के साथ कुछ कर लिया जाए, क्योंकि वह "प्रोसेस हो गया" जैसा लगता है। लेकिन अजनबी का मैसेज जानबूझकर नज़रअंदाज़ किया जाता है, इसलिए वह कभी कन्फ़र्म नहीं होता, इसलिए अगली कॉल पर तुरंत वापस आ जाता है, क्योंकि एक अपडेट बाकी है और Telegram के पास रिक्वेस्ट रोके रखने की कोई वजह नहीं।

Backoff दूसरी गड़बड़ी के लिए है: रद्द किया हुआ या गलत लिखा टोकन हर रिक्वेस्ट पर फ़ेल होता है, और पूरी रफ़्तार से दोबारा कोशिश करने का मतलब हमेशा के लिए हर कुछ सेकंड में एक 401। 5, 15, 60 और फिर 300 सेकंड की सीढ़ी वही है जो ऐप इस्तेमाल करता है।

टेस्ट में क्या दिखा

लोकल नकली सर्वर getUpdates (offset कन्फ़र्मेशन और timeout होल्ड) और sendMessage लागू करता है और हर कॉल रिकॉर्ड करता है। हर स्थिति में एक मैसेज डाला गया और फिर पढ़ा गया कि बॉट ने क्या भेजा।

फिर चार लूप 60 सेकंड चले, हर एक अपने नकली सर्वर के सामने:

नकली सर्वर पर, जो उसी मशीन से जवाब देता है, यह बग करीब 3,000 रिक्वेस्ट प्रति सेकंड पर चला। असली API के सामने लूप उतनी तेज़ चलेगा जितना नेटवर्क राउंड ट्रिप इजाज़त दे, हर कॉल पर उसी अजनबी का "hi" दोबारा आता हुआ और कुछ नया कभी न आता हुआ। जिसे आप कभी जवाब नहीं देंगे ऐसे किसी का एक मैसेज इसे शुरू करने के लिए काफ़ी है।

जो छोड़ा गया

असली ब्रिज को जवाबों को Telegram की 4,096 अक्षरों की सीमा के नीचे रखना भी पड़ता है (ऐप 3,900 पर काटता है, अक्षर की सीमा पर, क्योंकि बाइट के हिसाब से काटने पर जापानी अक्षर के बीच में कटाव आ सकता है), और जवाब सादे टेक्स्ट में भेजने पड़ते हैं। Markdown के साथ, अगर कोई तारा या अंडरस्कोर बेमेल हो तो Telegram पूरा मैसेज ठुकरा देता है, और snake_case कोड के बारे में लिखने वाला एजेंट ठीक यही पैदा करता है। सबसे ज़रूरी बात, जो टेक्स्ट आता है वह फ़ोन पकड़ने वाले का लिखा होता है, इसलिए उसे एजेंट तक अविश्वसनीय इनपुट की तरह पहुँचना चाहिए, ईमेल की तरह। Pairing तय करती है कि एजेंट से कौन बात कर सकता है; यह जो कहा गया है उसे सुरक्षित नहीं बना देती।

osFoundry डेस्कटॉप ऐप अपने AI एजेंट के लिए यह ब्रिज देता है: टोकन के लिए Device → Background, और किसी चैनल के Agents पैनल में "Pair a phone"। ऐप osFoundry डाउनलोड पेज से मुफ़्त है।