telegram-ctl

Remote-control and alert any long-running Python daemon from a Telegram chat. Register /commands with a decorator, get push alerts that never block your main loop, and a /help that writes itself. Zero dependencies — standard library only.

Your bot/scraper/pipeline is running on a server somewhere. You want to check on it and nudge it from your phone without SSHing in. That's this:

from telegram_ctl import Controller, Notifier

notify = Notifier()                 # reads TELEGRAM_TOKEN / TELEGRAM_CHAT_ID
ctl = Controller()

@ctl.command("status", help="current state")
def status(args):
    return f"📊 running — processed {worker.processed}"

@ctl.command("pause", help="stop taking new work")
async def pause(args):
    worker.paused = True
    return "⏸ paused"

notify.send("🚀 worker started")
await ctl.run(stop_event)           # long-polls Telegram, dispatches commands

From your chat: /status, /pause, /help … done.

Install

pip install -e .          # or just copy the telegram_ctl/ folder — no deps

Create a bot with @BotFather, then set:

export TELEGRAM_TOKEN=123456:ABC-your-token
export TELEGRAM_CHAT_ID=987654321      # your user/chat id

Why it's built this way (extracted from a 24/7 trading bot)

• Sends never block your loop. Notifier.send() dispatches on a daemon thread, so a slow Telegram round-trip can't stall your event loop — the exact failure mode that freezes naive bots.
• Only you can drive it. Commands from any chat id other than your TELEGRAM_CHAT_ID are silently ignored. No open command surface.
• Heartbeats don't spam. notify.throttled("hb", msg, every=3600) sends at most once an hour per key — for "still alive" pings and rate-limited warnings.
• Sync or async handlers. Return a string to reply, or None to stay quiet. The controller awaits coroutines automatically.
• /help writes itself from the help= text on each command.
• Testable offline. A RecordingTransport captures outgoing messages and feeds inbound ones, so the whole command flow is unit-tested with no network and no real bot (see tests/).

API

Notifier(token=None, chat_id=None, transport=None, parse_mode="HTML", block=False)
  .send(text)                       # fire-and-forget (sync if block=True)
  .throttled(key, text, every)      # True if sent, False if suppressed

Controller(token=None, chat_id=None, transport=None, poll_interval=2.0)
  @ctl.command(name, *aliases, help="")   # decorator; handler(args)->str|None
  ctl.reply(text)                         # push a message out-of-band
  await ctl.run(stop_event)               # poll + dispatch until stop_event set

Swap the transport (HTTPTransport by default) for RecordingTransport in tests, or your own class implementing send_message / get_updates.

Runnable example

examples/supervised_worker.py — a fake worker with /status /pause /resume /stop, event alerts, and a throttled heartbeat. Set the two env vars and run it.

Tests

python -m pytest -q       # 9 tests, no network required

Scope & disclaimer

Long-polling (getUpdates) — no webhook server, no inline keyboards, no file uploads; just commands + alerts, which is what a daemon usually needs. Not affiliated with Telegram. MIT licensed.