This is how to create a Telegram Bot including a small skeleton for a bot.
Create the Bot
Open your Telegram app and look for user @BotFather
Open a chat with him issue the command: /newbot
Now follow the instructions in Telegram and copy your token value
Install dependencies
put the following into a file named requirements.txt
requests
python-telegram-bot
tabulate
then run the following to install
pip install -r requirements.txt
The skeleton
#!/usr/bin/env python3.9
#
# Telegram Bot in python
# install modules as needed
#
import requests
import json
import logging
import os
import datetime
from telegram import Update, KeyboardButton, ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
from tabulate import tabulate
# BOT-Token from @BotFather
TOKEN = "<your Token>"
LOG_PATH = "/var/log/telegramBots/BotName.log"
# create log path
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
# set loglevel of httpx to warning
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(name)s | %(levelname)s | %(message)s",
encoding="utf-8",
handlers=[
logging.FileHandler(LOG_PATH),
logging.StreamHandler() # Konsole
]
)
logger = logging.getLogger(__name__)
# write PID
PID_FILE = "/run/telegramBots/telegram_bot.pid"
os.makedirs(os.path.dirname(PID_FILE), exist_ok=True)
with open(PID_FILE, 'w') as f:
f.write(str(os.getpid()))
f.close()
# handle errors somehow
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Log all errors.
"""
logger.error("Unerwarteter Fehler im Bot:", exc_info=context.error)
# Optional: Inform the User
if update and hasattr(update, "effective_message") and update.effective_message:
try:
await update.effective_message.reply_text(
"Unexpected server error. My Master has been notified!"
)
except Exception:
pass # Wenn sogar die Fehlermeldung nicht geht, einfach ignorieren
# you maybe want to restrict the bot or
# some commands only to dedicated user ids (like your own)
def isValidUser(update: Update, context: ContextTypes.DEFAULT_TYPE):
validUserList = ["<TelegramID_1","TelegramID_2"]
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
if str(sender.id) in validUserList:
logger.info(f"AUTHORIZED: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
return True
logger.warning(f"DENIED: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
return False
# The /start command
# This is always automatically issued if someone joins
# your bot, so, find some warm words...
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
await update.message.reply_text(
"Hello Master! This is your personal slave.\n"
"Throw some commands at me!\n"
"Use /help to get a list of things I can help you with.\n\n"
)
# Command /help
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE):
help_text = (
"Available commands:\n"
"/whoami - I'll tell you who you are\n"
"/ping - Echo Request\n"
"/help - Print this help text\n"
)
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
await update.message.reply_text(help_text)
# Command /whoami
async def cmd_whoami(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
text = (
f"📋 You are:\n"
f"ID: <code>{user.id}</code>\n"
f"Name: {user.full_name}\n"
f"Username: {user.username or '—'}\n"
f"IsBot: {'Yes' if user.is_bot else 'No'}\n"
f"Language: {user.language_code or 'unknown'}"
)
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
await update.message.reply_text(text, parse_mode='HTML')
# Command /ping
async def cmd_ping(update: Update, context: ContextTypes.DEFAULT_TYPE):
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
await update.message.reply_text("Pong! 🏓")
# main() function
def main():
app = Application.builder().token(TOKEN).build()
# Befehle registrieren
app.add_handler(CommandHandler("start", cmd_start))
app.add_handler(CommandHandler("help", cmd_help))
app.add_handler(CommandHandler("ping", cmd_ping))
app.add_handler(CommandHandler("whoami", cmd_whoami))
# Add more commands
# app.add_handler(CommandHandler("yourCommand", your_function))
# grab all locations that get sent to the bot
# that do not have a command in the message
#app.add_handler(MessageHandler(filters.LOCATION, handle_location))
# register the error_handler
app.add_error_handler(error_handler)
logger.info("Bot is starting...")
print("Bot is starting...")
app.run_polling()
if __name__ == '__main__':
main()
System integration
Systemd Service File
Minimal systemd service file, to run as deamon
[Unit]
Description=Telegram Bot
After=network.target
[Service]
WorkingDirectory=/tmp
ExecStart=/usr/local/bin/tlgrm_bot.py
ExecStop=/bin/kill -SIGINT $(cat /run/telegramBots/telegram_bot.pid)
Restart=always
RestartSec=10
User=bot_user
[Install]
WantedBy=multi-user.target
Logrotate
# cat /etc/logrotate.d/telegrambots
/var/log/telegramBots/*.log {
missingok
notifempty
daily
rotate 7
compress
delaycompress
copytruncate
}
Snippets
Create Command-Buttons
To create some command buttons we need to create a function first
def get_custom_keyboard():
keyboard = [
[
KeyboardButton("/getgas"),
KeyboardButton("/cryptoprice"),
KeyboardButton("/weather")
],
[
KeyboardButton("/whoami"),
KeyboardButton("/help")
]
]
return ReplyKeyboardMarkup(
keyboard,
resize_keyboard=True,
one_time_keyboard=False,
input_field_placeholder="Select a command..."
)
Optionally you can create a command that actually hides the buttons, if unwanted
async def cmd_hide(update: Update, context: ContextTypes.DEFAULT_TYPE):
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
await update.message.reply_text(
"Hiding shortcut buttons.\nSend /start again to get them back!",
reply_markup=ReplyKeyboardRemove()
)
Add the button keyboard within your /start command, which everyone issues first, talking to your bot
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
keyboard = get_custom_keyboard() # <-- HERE
await update.message.reply_text(
"Hello Master! This is your personal slave.\n"
"Throw some commands at me!\n"
"Use /help to get a list of things I can do for you.\n\n",
reply_markup=keyboard # <-- HERE
)
await cmd_help(update, context)
If you created the optional /hide command, add it to your main function and register the handler
def main():
...
app.add_handler(CommandHandler("hide", cmd_hide))
...
if __name__ == '__main__':
main()
Creating Commands that requires a Location
If you want to deliver localized results, the user needs to share his location with your bot.
Sadly, you can’t send a text message and attach location data to it. This must be handled in 2 messages, first the command e.g. /getmap and then the location message.
Happily, you can create context values within the message received and included in your response to the user. And in turn, if the user responds again, back to you.
There are several methods to achieve this, I personally decided for the following for my small bot.
So, what we need to achieve this, is a handler, that grabs each and every location message we receive and check these for our context flag.
GET_MAP_LOCATION = "waiting_for_getmap_location"
async def handle_location(update: Update, context: ContextTypes.DEFAULT_TYPE):
# only react if waiting for /getmap location
if context.user_data.get(GET_MAP_LOCATION):
# Reset value and call _cmd_getmap
context.user_data.pop(GET_MAP_LOCATION, None)
await _cmd_getmap(update, context)
else:
# pass if not waiting for location
# or send an error to the user
pass
def main():
...
app.add_handler(MessageHandler(filters.LOCATION, handle_location))
...
Now we need the function to handle the command itself and the one called by our location handler
async def cmd_getmap(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.message.location:
await _cmd_getmap(update, context)
return
# Save status of last message
# this will be read and reset by handle_location function
# on subsequent messages
context.user_data[GET_MAP_LOCATION] = True
# show 'Send Location' Button
keyboard = [
[KeyboardButton("Send Location", request_location=True)]
]
reply_markup = ReplyKeyboardMarkup(
keyboard,
resize_keyboard=True,
one_time_keyboard=True
)
sender = update.effective_user
username = f"@{sender.username}" if sender.username else "@Unknown"
logger.info(f"Command: {update.effective_message.text} for {sender.id} - {username} / {sender.full_name}")
await update.message.reply_text(
"Perfect! Just click the button below to let me know your location\n"
"and I will show you a Maps link!",
reply_markup=reply_markup
)
# gets called by handle_location function
# and only if context.user_data.pop(GET_MAP_LOCATION) has been set
async def _cmd_getmap(update: Update, context: ContextTypes.DEFAULT_TYPE):
# Build the link and respond
...
def main():
...
app.add_handler(CommandHandler("getmap", cmd_getmap))
...