I wanted a way to easily communicate with my agent from my phone; the easiest way to do this was via Telegram. Telegram also has a desktop application, which makes it easy to chat with on your computer.

Telegram has an official bot manager service called BotFather. BotFather is a first-party Telegram service that provides a single place to create and manage bots on Telegram. When you open Telegram, you search for @BotFather and give the command /newbot in the chat. BotFather will ask for a name for your bot, which you respond with, and just like that, you'll be issued an API token for your bot.

In order to use your API token within your code, you'll need to download the pyTelegramBotAPI.

pip install pyTelegramBotAPI

Once you've saved your API token to your .env you import telebot in your .py file then initialize your bot. 

import telebot

self.bot = telebot.TeleBot(self.token)

Processing Telegram Messages

When text is received, it's given to the LLM as input through a function called process_signal(). _build_signal_prompts() takes the message from the user and returns two strings: system_prompt and human_content. system_prompt and human_content are then given to _invoke_signal_llm_and_parse(). _apply_parsed_response_tags() completes the processing of the user's message and sends the response to the user. 

def process_signal(self, sender, subject, body):
    """Minimal: build prompts, invoke LLM, apply tags and send reply."""
    context = "(No memory yet.)"  # or "" if your prompt handles empty context
    system_prompt, human_content = self._build_signal_prompts(body, sender, context)

    try:
        res, res_content = self._invoke_signal_llm_and_parse(system_prompt, human_content, sender, body)
        self._apply_parsed_response_tags(res, sender, subject, body, None)  # no resonance
    except Exception as e:
        self.log(f"Signal processing error: {e}")
        if "Telegram" in sender:
            try:
                self.dispatch_as_human("Connection glitch — try again in a sec? 🦞")
            except Exception:
                pass

Checking Messages in Telegram

I set up a function called check_telegram() that runs every 5 seconds, checking for new messages.

There was an issue at one point where I was getting double messages due to some latency that would result in two instances of check_telegram() running at the same time. I resolved this by adding a True or False boolean. When check_telegram() runs, it checks if is_running is True; if so, it does not execute. If is_running is False, it changes to True and runs the check_telegram() function.

Every time a message is received from Telegram, it is assigned an ID by Telegram that you receive via the API. After processing the message, the ID is added to last_tg_update_id.txt. When a message is received from Telegram for processing, the ID is checked against the last_tg_update_id.txt file, and if the ID already exists, the message is not sent to the user.

_get_telegram_poll_offset() reads the last ID number that was saved in last_tg_update_id.txt so you're only given messages after that number. _fetch_telegram_updates() is what fetches the latest batch of messages following the highest number recorded. After messages are received, _mark_telegram_updates_seen() records the highest number so once again you can ask for only messages after that ID when the conversation continues.

_process_telegram_update() is what checks and validates the incoming message, e.g., is the message text or an image? Is the message coming from a verified chat? If the message was invalid, e.g., came from an unverified source, it would be ignored.

def check_telegram(self):
    if self.is_checking_tg:
        return
    self.is_checking_tg = True
    self.log("Checking Telegram updates...")
    try:
        offset = self._get_telegram_poll_offset()
        updates = self._fetch_telegram_updates(offset)
        if updates is None:
            return
        if updates:
            self.log(f"Updates found: {len(updates)}")
            self._mark_telegram_updates_seen(updates)
            for u in updates:
                self._process_telegram_update(u)
    except Exception as e:
        self.log(f"Telegram error: {e}")
    finally:
        self.is_checking_tg = False